Warning on dataproc while using partitionBy on a dataframe

Question

I am trying to write a dataframe contents to google cloud storage using a pyspark on dataproc. Though the write is successful, there are a lot of warning messages in the logs which I have pasted below. Is there some setting that I am missing that I need to while creating the cluster or in the pyspark program ? Or is this some google issue?

Note: Data written on the google storage by the dataframe is > 120 GB uncompressed. But I have noticed the same warnings even when I process data which is 1GB in size uncompressed. This is a simple dataframe with 50 columns that are read, some transformations are done and written to disk.

Dataframe Write statement is like below:

df.write.partitionBy("dt").format('csv').mode("overwrite").options(delimiter="|").save("gs://bucket/tbl/")

Warning statements in the Pyspark logs:

18/04/01 19:58:28 WARN org.apache.spark.scheduler.TaskSetManager: Lost task 182.0 in stage 3.0 (TID 68943, admg-tellrd-w-20.c.syw-analytics-repo-dev.internal, executor 219): org.apache.spark.SparkException: Task failed while writing rows
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:272)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala:191)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala:190)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
        at org.apache.spark.scheduler.Task.run(Task.scala:108)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:335)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalStateException: Error closing the output.
        at com.univocity.parsers.common.AbstractWriter.close(AbstractWriter.java:861)
        at org.apache.spark.sql.execution.datasources.csv.UnivocityGenerator.close(UnivocityGenerator.scala:86)
        at org.apache.spark.sql.execution.datasources.csv.CsvOutputWriter.close(CSVFileFormat.scala:141)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask.releaseResources(FileFormatWriter.scala:475)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask$$anonfun$execute$2.apply(FileFormatWriter.scala:450)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask$$anonfun$execute$2.apply(FileFormatWriter.scala:440)
        at scala.collection.Iterator$class.foreach(Iterator.scala:893)
        at org.apache.spark.sql.catalyst.util.AbstractScalaRowIterator.foreach(AbstractScalaRowIterator.scala:26)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask.execute(FileFormatWriter.scala:440)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:258)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:256)
        at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1375)
        at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:261)
        ... 8 more
Caused by: java.io.IOException: com.google.api.client.googleapis.json.GoogleJsonResponseException: 410 Gone
{
  "code" : 500,
  "errors" : [ {
    "domain" : "global",
    "message" : "Backend Error",
    "reason" : "backendError"
  } ],
  "message" : "Backend Error"
}
        at com.google.cloud.hadoop.util.AbstractGoogleAsyncWriteChannel.waitForCompletionAndThrowIfUploadFailed(AbstractGoogleAsyncWriteChannel.java:432)
        at com.google.cloud.hadoop.util.AbstractGoogleAsyncWriteChannel.close(AbstractGoogleAsyncWriteChannel.java:287)
        at java.nio.channels.Channels$1.close(Channels.java:178)
        at java.io.FilterOutputStream.close(FilterOutputStream.java:159)
        at com.google.cloud.hadoop.fs.gcs.GoogleHadoopOutputStream.close(GoogleHadoopOutputStream.java:126)
        at org.apache.hadoop.fs.FSDataOutputStream$PositionCache.close(FSDataOutputStream.java:72)
        at org.apache.hadoop.fs.FSDataOutputStream.close(FSDataOutputStream.java:106)
        at sun.nio.cs.StreamEncoder.implClose(StreamEncoder.java:320)
        at sun.nio.cs.StreamEncoder.close(StreamEncoder.java:149)
        at java.io.OutputStreamWriter.close(OutputStreamWriter.java:233)
        at com.univocity.parsers.common.AbstractWriter.close(AbstractWriter.java:857)
        ... 20 more
Caused by: com.google.api.client.googleapis.json.GoogleJsonResponseException: 410 Gone
{
  "code" : 500,
  "errors" : [ {
    "domain" : "global",
    "message" : "Backend Error",
    "reason" : "backendError"
  } ],
  "message" : "Backend Error"
}
        at com.google.api.client.googleapis.json.GoogleJsonResponseException.from(GoogleJsonResponseException.java:145)
        at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:113)
        at com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:40)
        at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:432)
        at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:352)
        at com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:469)
        at com.google.cloud.hadoop.util.AbstractGoogleAsyncWriteChannel$UploadOperation.call(AbstractGoogleAsyncWriteChannel.java:358)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        ... 3 more

Answer 1

I tried your code and it was indeed slow -- for me it took over 8 minutes.

I got a significant speedup (down to under 5 minutes) just by using Dataframes rather than RDDs to read the CSV files. This avoids shipping all the data between JVM <-> Python. Here's the code I used:

from pyspark.sql import *
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark import *

spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext

schema = schema = StructType([
  StructField("col1", StringType(), True),
  StructField("col2", StringType(), True),
  StructField("col3", StringType(), True),
  StructField("col4", StringType(), True),
  StructField("col5", StringType(), True)])

schema_df = spark.read.schema(schema).option("delimiter", "|").csv("gs://path/to/incremental_file.txt*")
schema_df.createOrReplaceTempView("df")

schema_df2 = spark.read.schema(schema).option("delimiter", "|").csv("gs://path/to/hist-file*.gz")
schema_df2.createOrReplaceTempView("df2")   

union_fn = schema_df2.union(schema_df)
w = Window.partitionBy("col1","col2").orderBy(col("col4").desc())
union_result = union_fn.withColumn("row_num", row_number().over(w)).where(col("row_num") == 1).drop("row_num").drop("col4")
union_result.createOrReplaceTempView("merged_tbl")
schema_merged_tbl = spark.sql("""select col1,col2,col3,col5 as col6 from merged_tbl""")
schema_merged_tbl.write.partitionBy("col6").format('csv').mode("overwrite").options(delimiter='|',codec="org.apache.hadoop.io.compress.GzipCodec").save("hdfs_merge_path")

Answer 2

(This is not an answer to the question, but this didn't fit in a comment. It's related to thread on calling repartition(...) before write.partitionBy).

Without repartition(...), that's going to take forever on GCS. Under the hood, when you say write.partitionBy(...), spark's tasks will each serially write one file at a time for each partition. That's already slow on HDFS, but since GCS has higher latency, that's going to be even slower. If it's 500ms to create each file, writing 2300 partition files will take ~20 minutes per task.

If you shuffle the data, you'll introduce another ("reduce") stage of tasks that will end up with all the data for one partition. So instead of writing 2300*previous-stage-task files, you'll just write 2300 files. This is what you want, especially when you have more than a few partitions.

You might want to experiment with the number of partitions that repartition(...) makes (aka reducer tasks). By default it is 200, but you may want to raise that. Each reducer will end up with a subset of the 2300 output partitions, and will serially write each output file. Again, assuming writing one file takes 500ms, 2300/200 = 115 files = ~1 minute per task. If you have more reducers, you'll get more parallelism, so each task will take less time. But you should set the number of reducers based on the number of nodes in your cluster (e.g. 4x the number of vcores).

Also, you might want to raise spark.executor.cores to 4 (--properties spark.executor.cores=4), since this will be pretty IO bound.

Answer 3

This is not the answer to the question but the code flow for the existing requirement.

col1        col2        col3    col4       col5
asd234qsds  2014-01-02  23.99  2014-01-02  Y
2343fsdf55  2014-01-03  22.56  2014-01-03  Y
123fdfr555  2014-01-04  34.23  2014-01-04  N
2343fsdf5f  2014-01-05  45.33  2014-01-05  N
asd234qsds  2014-01-02  27.99  2014-01-07  Y

Please note: the first and the last row have the same keys, but during the Window function only the last row would be considered. My actual data has 51 columns and the Window function is on 9 columns. Im not sure if the compressed data is adding any overhead to this process.

lines1 = sc.textFile("gs://incrmental_file.txt*") -- uncompressed data 210KB
part1 = lines1.map(lambda l: l.split("|")) 
df = part1.map(lambda c: Row(col1=c[0],col2=c[1],col3=c[2],col4=c[3], col5 =c[4]))
schema_df = spark.createDataFrame(df)
schema_df.createOrReplaceTempView("df")
#schema_incr_tbl = spark.sql("""select col1,col2,col3,col4,col5 from df""") 

lines2 = sc.textFile("gs://hist_files.gz*") -- full year compressed data 38GiB
part2 = lines2.map(lambda l: l.split("|"))
df2 = part2.map(lambda c: Row(col1=c[0],col2=c[1],col3=c[2],col4=c[3], col5 =c[4])) 
schema_df2 = spark.createDataFrame(df2)
schema_df2.createOrReplaceTempView("df2")   

union_fn = schema_hist_tbl.union(schema_incr_tbl)
w = Window.partitionBy("col1","col2").orderBy(col("col4").desc())
union_result = union_fn.withColumn("row_num", 
func.row_number().over(w)).where(col("row_num") == 1).drop("row_num").drop("col4")
union_result.createOrReplaceTempView("merged_tbl") 
schema_merged_tbl = spark.sql("""select col1,col2,col3,col5,col5 as col6 merged_tbl""") 

schema_merged_tbl.write.partitionBy("col6").format('csv').mode("overwrite").options(delimiter=delim,codec="org.apache.hadoop.io.compress.GzipCodec").save("hdfs_merge_path")