How to avoid reading old files from S3 when append

问题:

Once in 2 hours, spark job is running to convert some tgz files to parquet. The job appends the new data into an existing parquet in s3:

df.write.mode("append").partitionBy("id","day").parquet("s3://myBucket/foo.parquet")

In spark-submit output I can see significant time is being spent on reading old parquet files, for example:

16/11/27 14:06:15 INFO S3NativeFileSystem: Opening 's3://myBucket/foo.parquet/id=123/day=2016-11-26/part-r-00003-b20752e9-5d70-43f5-b8b4-50b5b4d0c7da.snappy.parquet' for reading

16/11/27 14:06:15 INFO S3NativeFileSystem: Stream for key 'foo.parquet/id=123/day=2016-11-26/part-r-00003-e80419de-7019-4859-bbe7-dcd392f6fcd3.snappy.parquet' seeking to position '149195444'

It looks like this operation takes less than 1 second per file, but the amount of files increases with time (each append adds new files), which makes me think that my code will not be able to scale.

Any ideas how to avoid reading old parquet files from s3 if I just need to append new data?

I use EMR 4.8.2 and DirectParquetOutputCommitter:

sc._jsc.hadoopConfiguration().set('spark.sql.parquet.output.committer.class', 'org.apache.spark.sql.parquet.DirectParquetOutputCommitter')

回答1:

I resolved this issue by writing the dataframe to EMR HDFS and then using s3-dist-cp uploading the parquets to S3

回答2:

Switch this over to using Dynamic Partition Overwrite Mode using:

.config("spark.sql.sources.partitionOverwriteMode", "dynamic")

Also, avoid the DirectParquetOutputCommitter, and instead don't modify this - you will achieve better results in terms of speed using the EMRFS File Committer.