{"action":"created","issue":{"url":"https://api.github.com/repos/aws-samples/aws-glue-samples/issues/8","repository_url":"https://api.github.com/repos/aws-samples/aws-glue-samples","labels_url":"https://api.github.com/repos/aws-samples/aws-glue-samples/issues/8/labels{/name}","comments_url":"https://api.github.com/repos/aws-samples/aws-glue-samples/issues/8/comments","events_url":"https://api.github.com/repos/aws-samples/aws-glue-samples/issues/8/events","html_url":"https://github.com/aws-samples/aws-glue-samples/issues/8","id":276523916,"node_id":"MDU6SXNzdWUyNzY1MjM5MTY=","number":8,"title":"AWS Glue job is failing for larger csv data on s3","user":{"login":"sumit1294","id":6001355,"node_id":"MDQ6VXNlcjYwMDEzNTU=","avatar_url":"https://avatars0.githubusercontent.com/u/6001355?v=4","gravatar_id":"","url":"https://api.github.com/users/sumit1294","html_url":"https://github.com/sumit1294","followers_url":"https://api.github.com/users/sumit1294/followers","following_url":"https://api.github.com/users/sumit1294/following{/other_user}","gists_url":"https://api.github.com/users/sumit1294/gists{/gist_id}","starred_url":"https://api.github.com/users/sumit1294/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sumit1294/subscriptions","organizations_url":"https://api.github.com/users/sumit1294/orgs","repos_url":"https://api.github.com/users/sumit1294/repos","events_url":"https://api.github.com/users/sumit1294/events{/privacy}","received_events_url":"https://api.github.com/users/sumit1294/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":5,"created_at":"2017-11-24T06:20:20Z","updated_at":"2018-06-06T01:57:45Z","closed_at":null,"author_association":"NONE","body":"For small s3 input files (~10GB), glue ETL job works fine but for the larger dataset (~200GB), the job is failing.\r\n\r\nAdding a part of ETL code.\r\n// Converting Dynamic frame to dataframe\r\ndf = dropnullfields3.toDF()\r\n// create new partition column\r\npartitioned_dataframe = df.withColumn('part_date', df['timestamp_utc'].cast('date'))\r\n// store the data in parquet format on s3 \r\npartitioned_dataframe.write.partitionBy(['part_date']).format(\"parquet\").save(output_lg_partitioned_dir, mode=\"append\")\r\n\r\nJob executed for 4 hours and threw an error.\r\n\r\nFile \"script_2017-11-23-15-07-32.py\", line 49, in <module>\r\npartitioned_dataframe.write.partitionBy(['part_date']).format(\"parquet\").save(output_lg_partitioned_dir, mode=\"append\")\r\nFile \"/mnt/yarn/usercache/root/appcache/application_1511449472652_0001/container_1511449472652_0001_02_000001/pyspark.zip/pyspark/sql/readwriter.py\", line 550, in save\r\nFile \"/mnt/yarn/usercache/root/appcache/application_1511449472652_0001/container_1511449472652_0001_02_000001/py4j-0.10.4-src.zip/py4j/java_gateway.py\", line 1133, in __call__\r\nFile \"/mnt/yarn/usercache/root/appcache/application_1511449472652_0001/container_1511449472652_0001_02_000001/pyspark.zip/pyspark/sql/utils.py\", line 63, in deco\r\nFile \"/mnt/yarn/usercache/root/appcache/application_1511449472652_0001/container_1511449472652_0001_02_000001/py4j-0.10.4-src.zip/py4j/protocol.py\", line 319, in get_return_value\r\npy4j.protocol.Py4JJavaError: An error occurred while calling o172.save.\r\n: org.apache.spark.SparkException: Job aborted.\r\nat org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply$mcV$sp(FileFormatWriter.scala:147)\r\nat org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:121)\r\nat org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:121)\r\nat org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:57)\r\nat org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:121)\r\nat org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:101)\r\nat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:58)\r\nat org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:56)\r\nat org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:74)\r\nat org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:114)\r\nat org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:114)\r\nat org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:135)\r\nat org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)\r\nat org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:132)\r\nat org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:113)\r\nat org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:87)\r\nat org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:87)\r\nat org.apache.spark.sql.execution.datasources.DataSource.write(DataSource.scala:492)\r\nat org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:215)\r\nat org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:198)\r\nat sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\r\nat sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\nat sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\nat java.lang.reflect.Method.invoke(Method.java:498)\r\nat py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)\r\nat py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)\r\nat py4j.Gateway.invoke(Gateway.java:280)\r\nat py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)\r\nat py4j.commands.CallCommand.execute(CallCommand.java:79)\r\nat py4j.GatewayConnection.run(GatewayConnection.java:214)\r\nat java.lang.Thread.run(Thread.java:748)\r\nCaused by: org.apache.spark.SparkException: Job aborted due to stage failure: Total size of serialized results of 3385 tasks (1024.1 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)\r\nat org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1435)\r\nat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1423)\r\nat org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1422)\r\nat scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)\r\nat scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)\r\nat org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1422)\r\nat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802)\r\nat org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:802)\r\nat scala.Option.foreach(Option.scala:257)\r\nat org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:802)\r\nat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1650)\r\nat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1605)\r\nat org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1594)\r\nat org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)\r\nat org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:628)\r\nat org.apache.spark.SparkContext.runJob(SparkContext.scala:1918)\r\nat org.apache.spark.SparkContext.runJob(SparkContext.scala:1931)\r\nat org.apache.spark.SparkContext.runJob(SparkContext.scala:1951)\r\nat org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply$mcV$sp(FileFormatWriter.scala:127)\r\n... 30 more\r\n\r\nEnd of LogType:stdout"},"comment":{"url":"https://api.github.com/repos/aws-samples/aws-glue-samples/issues/comments/394915076","html_url":"https://github.com/aws-samples/aws-glue-samples/issues/8#issuecomment-394915076","issue_url":"https://api.github.com/repos/aws-samples/aws-glue-samples/issues/8","id":394915076,"node_id":"MDEyOklzc3VlQ29tbWVudDM5NDkxNTA3Ng==","user":{"login":"mohitsax","id":30638067,"node_id":"MDQ6VXNlcjMwNjM4MDY3","avatar_url":"https://avatars2.githubusercontent.com/u/30638067?v=4","gravatar_id":"","url":"https://api.github.com/users/mohitsax","html_url":"https://github.com/mohitsax","followers_url":"https://api.github.com/users/mohitsax/followers","following_url":"https://api.github.com/users/mohitsax/following{/other_user}","gists_url":"https://api.github.com/users/mohitsax/gists{/gist_id}","starred_url":"https://api.github.com/users/mohitsax/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mohitsax/subscriptions","organizations_url":"https://api.github.com/users/mohitsax/orgs","repos_url":"https://api.github.com/users/mohitsax/repos","events_url":"https://api.github.com/users/mohitsax/events{/privacy}","received_events_url":"https://api.github.com/users/mohitsax/received_events","type":"User","site_admin":false},"created_at":"2018-06-06T01:57:44Z","updated_at":"2018-06-06T01:57:44Z","author_association":"COLLABORATOR","body":"The stack trace with the exception \"Total size of serialized results of 3385 tasks (1024.1 MB) is bigger than spark.driver.maxResultSize (1024.0 MB)\" indicates that the Spark driver is running OOM. This may happen in a variety of scenarios, such as: (1) Your job collect rdd at driver or broadcast large variables to executors, (2) you have a large number of input files (~10s of thousands) resulting in a large state on driver for keeping track of tasks processing each of those input files.\r\n\r\nIn your case, it seems that the job is processing 3385 or fewer CSV files, which should not ideally OOM out the driver. However, you may have another a prior stage in your job that may have resulted in a large number of tasks or resulted in large memory footprint for the driver. \r\n\r\nFor scenario 1, avoid collect'ing rdds at driver or large broadcast. For scenario 2, use Grouping feature in AWS Glue to read a large number of input files and enable Job Bookmarks to avoid re-processing old input data. \r\n\r\nMore documentation on how to use Grouping feature here:\r\nhttps://docs.aws.amazon.com/glue/latest/dg/grouping-input-files.html\r\n\r\nMore documentation on how to enable Job Bookmarks here:\r\nhttps://docs.aws.amazon.com/glue/latest/dg/monitor-continuations.html"}}
{"actor":{"display_login":"mohitsax"}}