[ https://issues.apache.org/jira/browse/SPARK-18819?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Apache Spark reassigned SPARK-18819: ------------------------------------ Assignee: Apache Spark > 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 > Assignee: Apache Spark > 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-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org