Question 102244 in Amazon Web Sevices Question posted in
The official Amazon Web Services documentation can be found here.

Amazon web services – Why does AWS EMR PySpark get stuck when I try to aggregate dataframe

I’m running a Spark application in AWS EMR. The code is like this:

with SparkSession.builder.appName(f"Spark App").getOrCreate() as spark:
    dataframe = spark.read.format('jdbc').options(
        ...
    ).load()
    print("Log A")
    max_date_result = dataframe.agg(max_(date_format('date', 'yyyy-MM-dd')).alias('max_date')).collect()[0]
    print("Log B")

This application always gets stuck for a long time before failing. According to the log printed, I can see Log A at the end of the log file, but no Log B. So I think the application was stuck by the aggregation. Is there any problem with the aggregation? Also, I’ve found some questions about this problem, they are stuck on calling collect() because they are not using SparkSession, but I’ve already been using SparkSession now, it still doesn’t work.

Here is some more log I found in EMR:

22/08/08 06:23:06 INFO JDBCRelation: Number of partitions: 11, WHERE clauses of these partitions: ...
22/08/08 06:23:07 INFO YarnSchedulerBackend$YarnDriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (172.31.9.57:58172) with ID 2,  ResourceProfileId 0
22/08/08 06:23:07 INFO ExecutorMonitor: New executor 2 has registered (new total is 1)
22/08/08 06:23:08 INFO BlockManagerMasterEndpoint: Registering block manager ip-***.us-west-1.compute.internal:37741 with 4.8 GiB RAM, BlockManagerId(2, ip-***.us-west-1.compute.internal, 37741, None)
22/08/08 06:23:10 INFO CodeGenerator: Code generated in 274.762854 ms
22/08/08 06:23:10 INFO DAGScheduler: Registering RDD 3 (collect at spark_rds_parquet.py:136) as input to shuffle 0
22/08/08 06:23:10 INFO DAGScheduler: Got map stage job 0 (collect at spark_rds_parquet.py:136) with 11 output partitions
22/08/08 06:23:10 INFO DAGScheduler: Final stage: ShuffleMapStage 0 (collect at spark_rds_parquet.py:136)
22/08/08 06:23:10 INFO DAGScheduler: Parents of final stage: List()
22/08/08 06:23:10 INFO DAGScheduler: Missing parents: List()
22/08/08 06:23:10 INFO DAGScheduler: Submitting ShuffleMapStage 0 (MapPartitionsRDD[3] at collect at spark_rds_parquet.py:136), which has no missing parents
22/08/08 06:23:10 INFO YarnAllocator: Driver requested a total number of 3 executor(s) for resource profile id: 0.
22/08/08 06:23:10 INFO YarnAllocator: Canceling requests for 47 executor container(s) to have a new desired total 3 executors.
22/08/08 06:23:10 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 23.8 KiB, free 1008.9 MiB)
22/08/08 06:23:10 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 11.6 KiB, free 1008.9 MiB)
22/08/08 06:23:10 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on ip-***.us-west-1.compute.internal:34379 (size: 11.6 KiB, free: 1008.9 MiB)
22/08/08 06:23:10 INFO SparkContext: Created broadcast 0 from broadcast at DAGScheduler.scala:1467
22/08/08 06:23:10 INFO DAGScheduler: Submitting 11 missing tasks from ShuffleMapStage 0 (MapPartitionsRDD[3] at collect at spark_rds_parquet.py:136) (first 15 tasks are for partitions Vector(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
22/08/08 06:23:10 INFO YarnClusterScheduler: Adding task set 0.0 with 11 tasks resource profile 0
22/08/08 06:23:10 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0) (ip-***.us-west-1.compute.internal, executor 2, partition 0, PROCESS_LOCAL, 4338 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID 1) (ip-***.us-west-1.compute.internal, executor 2, partition 1, PROCESS_LOCAL, 4342 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 2.0 in stage 0.0 (TID 2) (ip-***.us-west-1.compute.internal, executor 2, partition 2, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 3.0 in stage 0.0 (TID 3) (ip-***.us-west-1.compute.internal, executor 2, partition 3, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO YarnSchedulerBackend$YarnDriverEndpoint: Registered executor NettyRpcEndpointRef(spark-client://Executor) (172.31.10.76:57610) with ID 1,  ResourceProfileId 0
22/08/08 06:23:10 INFO ExecutorMonitor: New executor 1 has registered (new total is 3)
22/08/08 06:23:10 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on ip-***.us-west-1.compute.internal:37741 (size: 11.6 KiB, free: 4.8 GiB)
22/08/08 06:23:10 INFO BlockManagerMasterEndpoint: Registering block manager ip-***.us-west-1.compute.internal:46531 with 9.7 GiB RAM, BlockManagerId(1, ip-***.us-west-1.compute.internal, 46531, None)
22/08/08 06:23:10 INFO TaskSetManager: Starting task 4.0 in stage 0.0 (TID 4) (ip-***.us-west-1.compute.internal, executor 1, partition 4, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 5.0 in stage 0.0 (TID 5) (ip-***.us-west-1.compute.internal, executor 1, partition 5, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 6.0 in stage 0.0 (TID 6) (ip-***.us-west-1.compute.internal, executor 1, partition 6, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 7.0 in stage 0.0 (TID 7) (ip-***.us-west-1.compute.internal, executor 1, partition 7, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 8.0 in stage 0.0 (TID 8) (ip-***.us-west-1.compute.internal, executor 1, partition 8, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 9.0 in stage 0.0 (TID 9) (ip-***.us-west-1.compute.internal, executor 1, partition 9, PROCESS_LOCAL, 4343 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO TaskSetManager: Starting task 10.0 in stage 0.0 (TID 10) (ip-***.us-west-1.compute.internal, executor 1, partition 10, PROCESS_LOCAL, 4309 bytes) taskResourceAssignments Map()
22/08/08 06:23:10 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on ip-***.us-west-1.compute.internal:46531 (size: 11.6 KiB, free: 9.7 GiB)
22/08/08 06:27:46 INFO TaskSetManager: Finished task 3.0 in stage 0.0 (TID 3) in 275934 ms on ip-***.us-west-1.compute.internal (executor 2) (1/11)
22/08/08 06:27:46 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 275949 ms on ip-***.us-west-1.compute.internal (executor 2) (2/11)
22/08/08 06:27:46 INFO TaskSetManager: Finished task 2.0 in stage 0.0 (TID 2) in 276058 ms on ip-***.us-west-1.compute.internal (executor 2) (3/11)
22/08/08 06:27:46 INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID 1) in 276095 ms on ip-***.us-west-1.compute.internal (executor 2) (4/11)
22/08/08 06:27:46 INFO YarnAllocator: Driver requested a total number of 2 executor(s) for resource profile id: 0.
22/08/08 06:27:47 INFO TaskSetManager: Finished task 9.0 in stage 0.0 (TID 9) in 276952 ms on ip-***.us-west-1.compute.internal (executor 1) (5/11)
22/08/08 06:27:47 INFO TaskSetManager: Finished task 5.0 in stage 0.0 (TID 5) in 277031 ms on ip-***.us-west-1.compute.internal (executor 1) (6/11)
22/08/08 06:27:47 INFO TaskSetManager: Finished task 7.0 in stage 0.0 (TID 7) in 277186 ms on ip-***.us-west-1.compute.internal (executor 1) (7/11)
22/08/08 06:27:47 INFO TaskSetManager: Finished task 4.0 in stage 0.0 (TID 4) in 277264 ms on ip-***.us-west-1.compute.internal (executor 1) (8/11)
22/08/08 06:27:47 INFO YarnAllocator: Driver requested a total number of 1 executor(s) for resource profile id: 0.
22/08/08 06:27:48 INFO TaskSetManager: Finished task 8.0 in stage 0.0 (TID 8) in 277338 ms on ip-***.us-west-1.compute.internal (executor 1) (9/11)
22/08/08 06:27:48 INFO TaskSetManager: Finished task 6.0 in stage 0.0 (TID 6) in 277403 ms on ip-***.us-west-1.compute.internal (executor 1) (10/11)
22/08/08 06:28:46 INFO YarnClusterSchedulerBackend: Requesting to kill executor(s) 2
22/08/08 06:28:46 INFO YarnClusterSchedulerBackend: Actual list of executor(s) to be killed is 2
22/08/08 06:28:46 INFO ApplicationMaster$AMEndpoint: Driver requested to kill executor(s) 2.
22/08/08 06:28:46 INFO ExecutorAllocationManager: Executors 2 removed due to idle timeout.
22/08/08 06:28:48 INFO YarnSchedulerBackend$YarnDriverEndpoint: Disabling executor 2.
22/08/08 06:28:48 INFO DAGScheduler: Executor lost: 2 (epoch 0)
22/08/08 06:28:48 INFO YarnClusterScheduler: Executor 2 on ip-***.us-west-1.compute.internal killed by driver.
22/08/08 06:28:48 INFO ExecutorMonitor: Executor 2 is removed. Remove reason statistics: (gracefully decommissioned: 0, decommision unfinished: 0, driver killed: 1, unexpectedly exited: 0).
22/08/08 06:28:48 INFO BlockManagerMasterEndpoint: Trying to remove executor 2 from BlockManagerMaster.
22/08/08 06:28:48 INFO BlockManagerMasterEndpoint: Removing block manager BlockManagerId(2, ip-***.us-west-1.compute.internal, 37741, None)
22/08/08 06:28:48 INFO BlockManagerMaster: Removed 2 successfully in removeExecutor
Tags:

2

Answers


  1. Chosen as BEST ANSWER
    0

    I figure out that problem was because Spark's predicate pushdown, Spark pushdown the filter to database and cause deadlock in database.


  2. 0

    I believe it’s not related to the creation of SparkSession, it’s because your executor is being killed because of memory issue when it processes the specific task (partition 11). As some of your tasks are successful, I think you can try:

    1. Please check if there is data skewness in particular partition, which make your executor fail on that task.
    2. Increase the number of partition of your data.
    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search