Apache Spark on YARN: Large number of input data files (combine multiple input files in spark)

Question

A help for the implementation best practice is needed. The operating environment is as follows:

• Log data file arrives irregularly.
• The size of a log data file is from 3.9KB to 8.5MB. The average is about 1MB.
• The number of records of a data file is from 13 lines to 22000 lines. The average is about 2700 lines.
• Data file must be post-processed before aggregation.
• Post-processing algorithm can be changed.
• Post-processed file is managed separately with original data file, since the post-processing algorithm might be changed.
• Daily aggregation is performed. All post-processed data file must be filtered record-by-record and aggregation(average, max min…) is calculated.
• Since aggregation is fine-grained, the number of records after the aggregation is not so small. It can be about half of the number of the original records.
• At a point, the number of the post-processed file can be about 200,000.
• A data file should be able to be deleted individually.

In a test, I tried to process 160,000 post-processed files by Spark starting with sc.textFile() with glob path, it failed with OutOfMemory exception on the driver process.

What is the best practice to handle this kind of data? Should I use HBase instead of plain files to save post-processed data?

Answer 1

We wrote own loader. It solved our problem with small files in HDFS. It uses Hadoop CombineFileInputFormat. In our case it reduced the number of mappers from 100000 to approx 3000 and made job significantly faster.

https://github.com/RetailRocket/SparkMultiTool

Example:

import ru.retailrocket.spark.multitool.Loaders 
val sessions = Loaders.combineTextFile(sc, "file:///test/*") 
// or val sessions = Loaders.combineTextFile(sc, conf.weblogs(), size = 256, delim = "\n") 
// where size is split size in Megabytes, delim - line break character 
println(sessions.count())