You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Michael Kamprath (JIRA)" <ji...@apache.org> on 2016/12/25 00:40:58 UTC
[jira] [Updated] (SPARK-18819) Double alignment on ARM71 platform
[ https://issues.apache.org/jira/browse/SPARK-18819?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Michael Kamprath updated SPARK-18819:
-------------------------------------
Description:
_Note - Updated the ticket title to be reflective of what was found to be the underlying issue_
When I create a data frame in PySpark with a small row count (less than number executors), then write it to a parquet file, then load that parquet file into a new data frame, and finally do any sort of read against the loaded new data frame, Spark fails with an {{ExecutorLostFailure}}.
Example code to replicate this issue:
{code}
from pyspark.sql.types import *
rdd = sc.parallelize([('row1',1,4.33,'name'),('row2',2,3.14,'string')])
my_schema = StructType([
StructField("id", StringType(), True),
StructField("value1", IntegerType(), True),
StructField("value2", DoubleType(), True),
StructField("name",StringType(), True)
])
df = spark.createDataFrame( rdd, schema=my_schema)
df.write.parquet('hdfs://master:9000/user/michael/test_data',mode='overwrite')
newdf = spark.read.parquet('hdfs://master:9000/user/michael/test_data/')
newdf.take(1)
{code}
The error I get when the {{take}} step runs is:
{code}
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-2-a3aa06c0c511> in <module>()
1 newdf = spark.read.parquet('hdfs://master:9000/user/michael/test_data/')
----> 2 newdf.take(1)
/usr/local/spark/python/pyspark/sql/dataframe.py in take(self, num)
346 [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]
347 """
--> 348 return self.limit(num).collect()
349
350 @since(1.3)
/usr/local/spark/python/pyspark/sql/dataframe.py in collect(self)
308 """
309 with SCCallSiteSync(self._sc) as css:
--> 310 port = self._jdf.collectToPython()
311 return list(_load_from_socket(port, BatchedSerializer(PickleSerializer())))
312
/usr/local/spark/python/lib/py4j-0.10.3-src.zip/py4j/java_gateway.py in __call__(self, *args)
1131 answer = self.gateway_client.send_command(command)
1132 return_value = get_return_value(
-> 1133 answer, self.gateway_client, self.target_id, self.name)
1134
1135 for temp_arg in temp_args:
/usr/local/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/usr/local/spark/python/lib/py4j-0.10.3-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
317 raise Py4JJavaError(
318 "An error occurred while calling {0}{1}{2}.\n".
--> 319 format(target_id, ".", name), value)
320 else:
321 raise Py4JError(
Py4JJavaError: An error occurred while calling o54.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 4 times, most recent failure: Lost task 0.3 in stage 2.0 (TID 6, 10.10.10.4): ExecutorLostFailure (executor 2 exited caused by one of the running tasks) Reason: Remote RPC client disassociated. Likely due to containers exceeding thresholds, or network issues. Check driver logs for WARN messages.
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1454)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1442)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1441)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1441)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
at scala.Option.foreach(Option.scala:257)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:811)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1667)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1622)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1611)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:632)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1873)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1886)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1899)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:347)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:39)
at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply$mcI$sp(Dataset.scala:2526)
at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2523)
at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2523)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57)
at org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2546)
at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:2523)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:237)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:280)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:745)
{code}
The stdout logs of a failed executor contains:
{code}
#
# A fatal error has been detected by the Java Runtime Environment:
#
# SIGBUS (0x7) at pc=0xb68f92e0, pid=1424, tid=0x612ae460
#
# JRE version: Java(TM) SE Runtime Environment (8.0_101-b13) (build 1.8.0_101-b13)
# Java VM: Java HotSpot(TM) Client VM (25.101-b13 mixed mode linux-arm )
# Problematic frame:
# V [libjvm.so+0x4e72e0] Unsafe_GetDouble+0x6c
#
# Failed to write core dump. Core dumps have been disabled. To enable core dumping, try "ulimit -c unlimited" before starting Java again
#
# An error report file with more information is saved as:
# /opt/spark-2.0.2-bin-hadoop2.7/work/app-20161211093349-0000/3/hs_err_pid1424.log
{code}
While the stderr of a failed executor is:
{code}
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
16/12/11 09:33:51 INFO CoarseGrainedExecutorBackend: Started daemon with process name: 1424@slave2
16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for TERM
16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for HUP
16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for INT
16/12/11 09:33:54 INFO SecurityManager: Changing view acls to: hduser
16/12/11 09:33:54 INFO SecurityManager: Changing modify acls to: hduser
16/12/11 09:33:54 INFO SecurityManager: Changing view acls groups to:
16/12/11 09:33:54 INFO SecurityManager: Changing modify acls groups to:
16/12/11 09:33:54 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hduser); groups with view permissions: Set(); users with modify permissions: Set(hduser); groups with modify permissions: Set()
16/12/11 09:33:55 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:44389 after 342 ms (0 ms spent in bootstraps)
16/12/11 09:33:57 INFO SecurityManager: Changing view acls to: hduser
16/12/11 09:33:57 INFO SecurityManager: Changing modify acls to: hduser
16/12/11 09:33:57 INFO SecurityManager: Changing view acls groups to:
16/12/11 09:33:57 INFO SecurityManager: Changing modify acls groups to:
16/12/11 09:33:57 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hduser); groups with view permissions: Set(); users with modify permissions: Set(hduser); groups with modify permissions: Set()
16/12/11 09:33:57 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:44389 after 15 ms (0 ms spent in bootstraps)
16/12/11 09:33:58 INFO DiskBlockManager: Created local directory at /data/spark/spark-161cf7dc-377b-4f40-94d9-b1928f124966/executor-517734a6-11d3-4ad1-94a0-cf5642a0ff22/blockmgr-dbef9ae3-3249-4455-8eec-3dae57798c8c
16/12/11 09:33:58 INFO MemoryStore: MemoryStore started with capacity 516.0 MB
16/12/11 09:33:58 INFO CoarseGrainedExecutorBackend: Connecting to driver: spark://CoarseGrainedScheduler@10.10.10.1:44389
16/12/11 09:33:58 INFO WorkerWatcher: Connecting to worker spark://Worker@10.10.10.3:45672
16/12/11 09:33:58 INFO TransportClientFactory: Successfully created connection to /10.10.10.3:45672 after 9 ms (0 ms spent in bootstraps)
16/12/11 09:33:59 INFO WorkerWatcher: Successfully connected to spark://Worker@10.10.10.3:45672
16/12/11 09:33:59 INFO CoarseGrainedExecutorBackend: Successfully registered with driver
16/12/11 09:33:59 INFO Executor: Starting executor ID 3 on host 10.10.10.3
16/12/11 09:33:59 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 43844.
16/12/11 09:33:59 INFO NettyBlockTransferService: Server created on 10.10.10.3:43844
16/12/11 09:33:59 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(3, 10.10.10.3, 43844)
16/12/11 09:33:59 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(3, 10.10.10.3, 43844)
16/12/11 09:34:44 INFO CoarseGrainedExecutorBackend: Got assigned task 2
16/12/11 09:34:44 INFO Executor: Running task 0.0 in stage 1.0 (TID 2)
16/12/11 09:34:45 INFO TorrentBroadcast: Started reading broadcast variable 1
16/12/11 09:34:45 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:37106 after 5 ms (0 ms spent in bootstraps)
16/12/11 09:34:45 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 25.8 KB, free 516.0 MB)
16/12/11 09:34:46 INFO TorrentBroadcast: Reading broadcast variable 1 took 543 ms
16/12/11 09:34:46 WARN SizeEstimator: Failed to check whether UseCompressedOops is set; assuming yes
16/12/11 09:34:46 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 71.4 KB, free 515.9 MB)
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
16/12/11 09:34:50 INFO Executor: Finished task 0.0 in stage 1.0 (TID 2). 2135 bytes result sent to driver
16/12/11 09:35:03 INFO CoarseGrainedExecutorBackend: Got assigned task 4
16/12/11 09:35:03 INFO Executor: Running task 0.1 in stage 2.0 (TID 4)
16/12/11 09:35:03 INFO TorrentBroadcast: Started reading broadcast variable 3
16/12/11 09:35:03 INFO MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 4.4 KB, free 516.0 MB)
16/12/11 09:35:03 INFO TorrentBroadcast: Reading broadcast variable 3 took 102 ms
16/12/11 09:35:03 INFO MemoryStore: Block broadcast_3 stored as values in memory (estimated size 9.0 KB, free 516.0 MB)
16/12/11 09:35:05 INFO CodeGenerator: Code generated in 958.630042 ms
16/12/11 09:35:05 INFO FileScanRDD: Reading File path: hdfs://master:9000/user/michael/test_data/part-r-00001-b802e900-dfaa-4fb7-aa2f-fb07d122d033.snappy.parquet, range: 0-889, partition values: [empty row]
16/12/11 09:35:05 INFO TorrentBroadcast: Started reading broadcast variable 2
16/12/11 09:35:05 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 24.9 KB, free 516.0 MB)
16/12/11 09:35:05 INFO TorrentBroadcast: Reading broadcast variable 2 took 57 ms
16/12/11 09:35:05 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 349.5 KB, free 515.6 MB)
16/12/11 09:35:05 INFO CodecPool: Got brand-new decompressor [.snappy]
{code}
I have tested this against HDFS 2.7 and QFS 1.2 on an ARM v7.1 based cluster. Both have the same results. Note I have verified this issue doesn't express on x86 platforms. The java version installed is Oracle's 1.8.0_101.
I generally discovered this when processing larger files that have individual parquet part files with a single row in them. The same problem manifested then.
was:
When I create a data frame in PySpark with a small row count (less than number executors), then write it to a parquet file, then load that parquet file into a new data frame, and finally do any sort of read against the loaded new data frame, Spark fails with an {{ExecutorLostFailure}}.
Example code to replicate this issue:
{code}
from pyspark.sql.types import *
rdd = sc.parallelize([('row1',1,4.33,'name'),('row2',2,3.14,'string')])
my_schema = StructType([
StructField("id", StringType(), True),
StructField("value1", IntegerType(), True),
StructField("value2", DoubleType(), True),
StructField("name",StringType(), True)
])
df = spark.createDataFrame( rdd, schema=my_schema)
df.write.parquet('hdfs://master:9000/user/michael/test_data',mode='overwrite')
newdf = spark.read.parquet('hdfs://master:9000/user/michael/test_data/')
newdf.take(1)
{code}
The error I get when the {{take}} step runs is:
{code}
---------------------------------------------------------------------------
Py4JJavaError Traceback (most recent call last)
<ipython-input-2-a3aa06c0c511> in <module>()
1 newdf = spark.read.parquet('hdfs://master:9000/user/michael/test_data/')
----> 2 newdf.take(1)
/usr/local/spark/python/pyspark/sql/dataframe.py in take(self, num)
346 [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]
347 """
--> 348 return self.limit(num).collect()
349
350 @since(1.3)
/usr/local/spark/python/pyspark/sql/dataframe.py in collect(self)
308 """
309 with SCCallSiteSync(self._sc) as css:
--> 310 port = self._jdf.collectToPython()
311 return list(_load_from_socket(port, BatchedSerializer(PickleSerializer())))
312
/usr/local/spark/python/lib/py4j-0.10.3-src.zip/py4j/java_gateway.py in __call__(self, *args)
1131 answer = self.gateway_client.send_command(command)
1132 return_value = get_return_value(
-> 1133 answer, self.gateway_client, self.target_id, self.name)
1134
1135 for temp_arg in temp_args:
/usr/local/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
61 def deco(*a, **kw):
62 try:
---> 63 return f(*a, **kw)
64 except py4j.protocol.Py4JJavaError as e:
65 s = e.java_exception.toString()
/usr/local/spark/python/lib/py4j-0.10.3-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
317 raise Py4JJavaError(
318 "An error occurred while calling {0}{1}{2}.\n".
--> 319 format(target_id, ".", name), value)
320 else:
321 raise Py4JError(
Py4JJavaError: An error occurred while calling o54.collectToPython.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 4 times, most recent failure: Lost task 0.3 in stage 2.0 (TID 6, 10.10.10.4): ExecutorLostFailure (executor 2 exited caused by one of the running tasks) Reason: Remote RPC client disassociated. Likely due to containers exceeding thresholds, or network issues. Check driver logs for WARN messages.
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1454)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1442)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1441)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1441)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
at scala.Option.foreach(Option.scala:257)
at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:811)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1667)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1622)
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1611)
at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:632)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1873)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1886)
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1899)
at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:347)
at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:39)
at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply$mcI$sp(Dataset.scala:2526)
at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2523)
at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2523)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57)
at org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2546)
at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:2523)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:237)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:280)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:214)
at java.lang.Thread.run(Thread.java:745)
{code}
The stdout logs of a failed executor contains:
{code}
#
# A fatal error has been detected by the Java Runtime Environment:
#
# SIGBUS (0x7) at pc=0xb68f92e0, pid=1424, tid=0x612ae460
#
# JRE version: Java(TM) SE Runtime Environment (8.0_101-b13) (build 1.8.0_101-b13)
# Java VM: Java HotSpot(TM) Client VM (25.101-b13 mixed mode linux-arm )
# Problematic frame:
# V [libjvm.so+0x4e72e0] Unsafe_GetDouble+0x6c
#
# Failed to write core dump. Core dumps have been disabled. To enable core dumping, try "ulimit -c unlimited" before starting Java again
#
# An error report file with more information is saved as:
# /opt/spark-2.0.2-bin-hadoop2.7/work/app-20161211093349-0000/3/hs_err_pid1424.log
{code}
While the stderr of a failed executor is:
{code}
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
16/12/11 09:33:51 INFO CoarseGrainedExecutorBackend: Started daemon with process name: 1424@slave2
16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for TERM
16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for HUP
16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for INT
16/12/11 09:33:54 INFO SecurityManager: Changing view acls to: hduser
16/12/11 09:33:54 INFO SecurityManager: Changing modify acls to: hduser
16/12/11 09:33:54 INFO SecurityManager: Changing view acls groups to:
16/12/11 09:33:54 INFO SecurityManager: Changing modify acls groups to:
16/12/11 09:33:54 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hduser); groups with view permissions: Set(); users with modify permissions: Set(hduser); groups with modify permissions: Set()
16/12/11 09:33:55 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:44389 after 342 ms (0 ms spent in bootstraps)
16/12/11 09:33:57 INFO SecurityManager: Changing view acls to: hduser
16/12/11 09:33:57 INFO SecurityManager: Changing modify acls to: hduser
16/12/11 09:33:57 INFO SecurityManager: Changing view acls groups to:
16/12/11 09:33:57 INFO SecurityManager: Changing modify acls groups to:
16/12/11 09:33:57 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hduser); groups with view permissions: Set(); users with modify permissions: Set(hduser); groups with modify permissions: Set()
16/12/11 09:33:57 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:44389 after 15 ms (0 ms spent in bootstraps)
16/12/11 09:33:58 INFO DiskBlockManager: Created local directory at /data/spark/spark-161cf7dc-377b-4f40-94d9-b1928f124966/executor-517734a6-11d3-4ad1-94a0-cf5642a0ff22/blockmgr-dbef9ae3-3249-4455-8eec-3dae57798c8c
16/12/11 09:33:58 INFO MemoryStore: MemoryStore started with capacity 516.0 MB
16/12/11 09:33:58 INFO CoarseGrainedExecutorBackend: Connecting to driver: spark://CoarseGrainedScheduler@10.10.10.1:44389
16/12/11 09:33:58 INFO WorkerWatcher: Connecting to worker spark://Worker@10.10.10.3:45672
16/12/11 09:33:58 INFO TransportClientFactory: Successfully created connection to /10.10.10.3:45672 after 9 ms (0 ms spent in bootstraps)
16/12/11 09:33:59 INFO WorkerWatcher: Successfully connected to spark://Worker@10.10.10.3:45672
16/12/11 09:33:59 INFO CoarseGrainedExecutorBackend: Successfully registered with driver
16/12/11 09:33:59 INFO Executor: Starting executor ID 3 on host 10.10.10.3
16/12/11 09:33:59 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 43844.
16/12/11 09:33:59 INFO NettyBlockTransferService: Server created on 10.10.10.3:43844
16/12/11 09:33:59 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(3, 10.10.10.3, 43844)
16/12/11 09:33:59 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(3, 10.10.10.3, 43844)
16/12/11 09:34:44 INFO CoarseGrainedExecutorBackend: Got assigned task 2
16/12/11 09:34:44 INFO Executor: Running task 0.0 in stage 1.0 (TID 2)
16/12/11 09:34:45 INFO TorrentBroadcast: Started reading broadcast variable 1
16/12/11 09:34:45 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:37106 after 5 ms (0 ms spent in bootstraps)
16/12/11 09:34:45 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 25.8 KB, free 516.0 MB)
16/12/11 09:34:46 INFO TorrentBroadcast: Reading broadcast variable 1 took 543 ms
16/12/11 09:34:46 WARN SizeEstimator: Failed to check whether UseCompressedOops is set; assuming yes
16/12/11 09:34:46 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 71.4 KB, free 515.9 MB)
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
16/12/11 09:34:50 INFO Executor: Finished task 0.0 in stage 1.0 (TID 2). 2135 bytes result sent to driver
16/12/11 09:35:03 INFO CoarseGrainedExecutorBackend: Got assigned task 4
16/12/11 09:35:03 INFO Executor: Running task 0.1 in stage 2.0 (TID 4)
16/12/11 09:35:03 INFO TorrentBroadcast: Started reading broadcast variable 3
16/12/11 09:35:03 INFO MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 4.4 KB, free 516.0 MB)
16/12/11 09:35:03 INFO TorrentBroadcast: Reading broadcast variable 3 took 102 ms
16/12/11 09:35:03 INFO MemoryStore: Block broadcast_3 stored as values in memory (estimated size 9.0 KB, free 516.0 MB)
16/12/11 09:35:05 INFO CodeGenerator: Code generated in 958.630042 ms
16/12/11 09:35:05 INFO FileScanRDD: Reading File path: hdfs://master:9000/user/michael/test_data/part-r-00001-b802e900-dfaa-4fb7-aa2f-fb07d122d033.snappy.parquet, range: 0-889, partition values: [empty row]
16/12/11 09:35:05 INFO TorrentBroadcast: Started reading broadcast variable 2
16/12/11 09:35:05 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 24.9 KB, free 516.0 MB)
16/12/11 09:35:05 INFO TorrentBroadcast: Reading broadcast variable 2 took 57 ms
16/12/11 09:35:05 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 349.5 KB, free 515.6 MB)
16/12/11 09:35:05 INFO CodecPool: Got brand-new decompressor [.snappy]
{code}
I have tested this against HDFS 2.7 and QFS 1.2 on an ARM v7.1 based cluster. Both have the same results. Note I have verified this issue doesn't express on x86 platforms. The java version installed is Oracle's 1.8.0_101.
I generally discovered this when processing larger files that have individual parquet part files with a single row in them. The same problem manifested then.
Summary: Double alignment on ARM71 platform (was: Failure to read single-row Parquet files)
> Double alignment on ARM71 platform
> ----------------------------------
>
> Key: SPARK-18819
> URL: https://issues.apache.org/jira/browse/SPARK-18819
> Project: Spark
> Issue Type: Bug
> Components: Input/Output, PySpark
> Affects Versions: 2.0.2
> Environment: Ubuntu 14.04 LTS on ARM 7.1
> Reporter: Michael Kamprath
> Priority: Critical
>
> _Note - Updated the ticket title to be reflective of what was found to be the underlying issue_
> When I create a data frame in PySpark with a small row count (less than number executors), then write it to a parquet file, then load that parquet file into a new data frame, and finally do any sort of read against the loaded new data frame, Spark fails with an {{ExecutorLostFailure}}.
> Example code to replicate this issue:
> {code}
> from pyspark.sql.types import *
> rdd = sc.parallelize([('row1',1,4.33,'name'),('row2',2,3.14,'string')])
> my_schema = StructType([
> StructField("id", StringType(), True),
> StructField("value1", IntegerType(), True),
> StructField("value2", DoubleType(), True),
> StructField("name",StringType(), True)
> ])
> df = spark.createDataFrame( rdd, schema=my_schema)
> df.write.parquet('hdfs://master:9000/user/michael/test_data',mode='overwrite')
> newdf = spark.read.parquet('hdfs://master:9000/user/michael/test_data/')
> newdf.take(1)
> {code}
> The error I get when the {{take}} step runs is:
> {code}
> ---------------------------------------------------------------------------
> Py4JJavaError Traceback (most recent call last)
> <ipython-input-2-a3aa06c0c511> in <module>()
> 1 newdf = spark.read.parquet('hdfs://master:9000/user/michael/test_data/')
> ----> 2 newdf.take(1)
> /usr/local/spark/python/pyspark/sql/dataframe.py in take(self, num)
> 346 [Row(age=2, name=u'Alice'), Row(age=5, name=u'Bob')]
> 347 """
> --> 348 return self.limit(num).collect()
> 349
> 350 @since(1.3)
> /usr/local/spark/python/pyspark/sql/dataframe.py in collect(self)
> 308 """
> 309 with SCCallSiteSync(self._sc) as css:
> --> 310 port = self._jdf.collectToPython()
> 311 return list(_load_from_socket(port, BatchedSerializer(PickleSerializer())))
> 312
> /usr/local/spark/python/lib/py4j-0.10.3-src.zip/py4j/java_gateway.py in __call__(self, *args)
> 1131 answer = self.gateway_client.send_command(command)
> 1132 return_value = get_return_value(
> -> 1133 answer, self.gateway_client, self.target_id, self.name)
> 1134
> 1135 for temp_arg in temp_args:
> /usr/local/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
> 61 def deco(*a, **kw):
> 62 try:
> ---> 63 return f(*a, **kw)
> 64 except py4j.protocol.Py4JJavaError as e:
> 65 s = e.java_exception.toString()
> /usr/local/spark/python/lib/py4j-0.10.3-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
> 317 raise Py4JJavaError(
> 318 "An error occurred while calling {0}{1}{2}.\n".
> --> 319 format(target_id, ".", name), value)
> 320 else:
> 321 raise Py4JError(
> Py4JJavaError: An error occurred while calling o54.collectToPython.
> : org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 2.0 failed 4 times, most recent failure: Lost task 0.3 in stage 2.0 (TID 6, 10.10.10.4): ExecutorLostFailure (executor 2 exited caused by one of the running tasks) Reason: Remote RPC client disassociated. Likely due to containers exceeding thresholds, or network issues. Check driver logs for WARN messages.
> Driver stacktrace:
> at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1454)
> at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1442)
> at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1441)
> at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
> at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1441)
> at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
> at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811)
> at scala.Option.foreach(Option.scala:257)
> at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:811)
> at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1667)
> at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1622)
> at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1611)
> at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
> at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:632)
> at org.apache.spark.SparkContext.runJob(SparkContext.scala:1873)
> at org.apache.spark.SparkContext.runJob(SparkContext.scala:1886)
> at org.apache.spark.SparkContext.runJob(SparkContext.scala:1899)
> at org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:347)
> at org.apache.spark.sql.execution.CollectLimitExec.executeCollect(limit.scala:39)
> at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply$mcI$sp(Dataset.scala:2526)
> at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2523)
> at org.apache.spark.sql.Dataset$$anonfun$collectToPython$1.apply(Dataset.scala:2523)
> at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57)
> at org.apache.spark.sql.Dataset.withNewExecutionId(Dataset.scala:2546)
> at org.apache.spark.sql.Dataset.collectToPython(Dataset.scala:2523)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:237)
> at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
> at py4j.Gateway.invoke(Gateway.java:280)
> at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
> at py4j.commands.CallCommand.execute(CallCommand.java:79)
> at py4j.GatewayConnection.run(GatewayConnection.java:214)
> at java.lang.Thread.run(Thread.java:745)
> {code}
> The stdout logs of a failed executor contains:
> {code}
> #
> # A fatal error has been detected by the Java Runtime Environment:
> #
> # SIGBUS (0x7) at pc=0xb68f92e0, pid=1424, tid=0x612ae460
> #
> # JRE version: Java(TM) SE Runtime Environment (8.0_101-b13) (build 1.8.0_101-b13)
> # Java VM: Java HotSpot(TM) Client VM (25.101-b13 mixed mode linux-arm )
> # Problematic frame:
> # V [libjvm.so+0x4e72e0] Unsafe_GetDouble+0x6c
> #
> # Failed to write core dump. Core dumps have been disabled. To enable core dumping, try "ulimit -c unlimited" before starting Java again
> #
> # An error report file with more information is saved as:
> # /opt/spark-2.0.2-bin-hadoop2.7/work/app-20161211093349-0000/3/hs_err_pid1424.log
> {code}
> While the stderr of a failed executor is:
> {code}
> Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
> 16/12/11 09:33:51 INFO CoarseGrainedExecutorBackend: Started daemon with process name: 1424@slave2
> 16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for TERM
> 16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for HUP
> 16/12/11 09:33:51 INFO SignalUtils: Registered signal handler for INT
> 16/12/11 09:33:54 INFO SecurityManager: Changing view acls to: hduser
> 16/12/11 09:33:54 INFO SecurityManager: Changing modify acls to: hduser
> 16/12/11 09:33:54 INFO SecurityManager: Changing view acls groups to:
> 16/12/11 09:33:54 INFO SecurityManager: Changing modify acls groups to:
> 16/12/11 09:33:54 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hduser); groups with view permissions: Set(); users with modify permissions: Set(hduser); groups with modify permissions: Set()
> 16/12/11 09:33:55 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:44389 after 342 ms (0 ms spent in bootstraps)
> 16/12/11 09:33:57 INFO SecurityManager: Changing view acls to: hduser
> 16/12/11 09:33:57 INFO SecurityManager: Changing modify acls to: hduser
> 16/12/11 09:33:57 INFO SecurityManager: Changing view acls groups to:
> 16/12/11 09:33:57 INFO SecurityManager: Changing modify acls groups to:
> 16/12/11 09:33:57 INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(hduser); groups with view permissions: Set(); users with modify permissions: Set(hduser); groups with modify permissions: Set()
> 16/12/11 09:33:57 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:44389 after 15 ms (0 ms spent in bootstraps)
> 16/12/11 09:33:58 INFO DiskBlockManager: Created local directory at /data/spark/spark-161cf7dc-377b-4f40-94d9-b1928f124966/executor-517734a6-11d3-4ad1-94a0-cf5642a0ff22/blockmgr-dbef9ae3-3249-4455-8eec-3dae57798c8c
> 16/12/11 09:33:58 INFO MemoryStore: MemoryStore started with capacity 516.0 MB
> 16/12/11 09:33:58 INFO CoarseGrainedExecutorBackend: Connecting to driver: spark://CoarseGrainedScheduler@10.10.10.1:44389
> 16/12/11 09:33:58 INFO WorkerWatcher: Connecting to worker spark://Worker@10.10.10.3:45672
> 16/12/11 09:33:58 INFO TransportClientFactory: Successfully created connection to /10.10.10.3:45672 after 9 ms (0 ms spent in bootstraps)
> 16/12/11 09:33:59 INFO WorkerWatcher: Successfully connected to spark://Worker@10.10.10.3:45672
> 16/12/11 09:33:59 INFO CoarseGrainedExecutorBackend: Successfully registered with driver
> 16/12/11 09:33:59 INFO Executor: Starting executor ID 3 on host 10.10.10.3
> 16/12/11 09:33:59 INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port 43844.
> 16/12/11 09:33:59 INFO NettyBlockTransferService: Server created on 10.10.10.3:43844
> 16/12/11 09:33:59 INFO BlockManagerMaster: Registering BlockManager BlockManagerId(3, 10.10.10.3, 43844)
> 16/12/11 09:33:59 INFO BlockManagerMaster: Registered BlockManager BlockManagerId(3, 10.10.10.3, 43844)
> 16/12/11 09:34:44 INFO CoarseGrainedExecutorBackend: Got assigned task 2
> 16/12/11 09:34:44 INFO Executor: Running task 0.0 in stage 1.0 (TID 2)
> 16/12/11 09:34:45 INFO TorrentBroadcast: Started reading broadcast variable 1
> 16/12/11 09:34:45 INFO TransportClientFactory: Successfully created connection to /10.10.10.1:37106 after 5 ms (0 ms spent in bootstraps)
> 16/12/11 09:34:45 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 25.8 KB, free 516.0 MB)
> 16/12/11 09:34:46 INFO TorrentBroadcast: Reading broadcast variable 1 took 543 ms
> 16/12/11 09:34:46 WARN SizeEstimator: Failed to check whether UseCompressedOops is set; assuming yes
> 16/12/11 09:34:46 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 71.4 KB, free 515.9 MB)
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
> 16/12/11 09:34:50 INFO Executor: Finished task 0.0 in stage 1.0 (TID 2). 2135 bytes result sent to driver
> 16/12/11 09:35:03 INFO CoarseGrainedExecutorBackend: Got assigned task 4
> 16/12/11 09:35:03 INFO Executor: Running task 0.1 in stage 2.0 (TID 4)
> 16/12/11 09:35:03 INFO TorrentBroadcast: Started reading broadcast variable 3
> 16/12/11 09:35:03 INFO MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 4.4 KB, free 516.0 MB)
> 16/12/11 09:35:03 INFO TorrentBroadcast: Reading broadcast variable 3 took 102 ms
> 16/12/11 09:35:03 INFO MemoryStore: Block broadcast_3 stored as values in memory (estimated size 9.0 KB, free 516.0 MB)
> 16/12/11 09:35:05 INFO CodeGenerator: Code generated in 958.630042 ms
> 16/12/11 09:35:05 INFO FileScanRDD: Reading File path: hdfs://master:9000/user/michael/test_data/part-r-00001-b802e900-dfaa-4fb7-aa2f-fb07d122d033.snappy.parquet, range: 0-889, partition values: [empty row]
> 16/12/11 09:35:05 INFO TorrentBroadcast: Started reading broadcast variable 2
> 16/12/11 09:35:05 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 24.9 KB, free 516.0 MB)
> 16/12/11 09:35:05 INFO TorrentBroadcast: Reading broadcast variable 2 took 57 ms
> 16/12/11 09:35:05 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 349.5 KB, free 515.6 MB)
> 16/12/11 09:35:05 INFO CodecPool: Got brand-new decompressor [.snappy]
> {code}
> I have tested this against HDFS 2.7 and QFS 1.2 on an ARM v7.1 based cluster. Both have the same results. Note I have verified this issue doesn't express on x86 platforms. The java version installed is Oracle's 1.8.0_101.
> I generally discovered this when processing larger files that have individual parquet part files with a single row in them. The same problem manifested then.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org