Spark Streaming读写Kafka后写入HDFS-Hive：如何控制文件行数与大小？

Absolutely! Dealing with tiny files in HDFS/Hive when using Spark Streaming + Kafka is such a common headache—we’ve all been there. The good news is you absolutely can enforce minimum file sizes or row counts before writing, with timeouts as a safety net to avoid hanging indefinitely. Let’s walk through the practical approaches:

For both DStream and Structured Streaming, Spark has built-in settings to control file output sizes. Combine these to avoid tiny files:

• spark.sql.files.minPartitionBytes: Sets the minimum size for each partition (and thus each output file, assuming 1 file per partition). For example, setting this to 134217728 (128MB) ensures each file is at least 128MB before being written.
• spark.sql.streaming.fileSink.rollingInterval: For Structured Streaming, this controls how often the file sink rolls over to a new file. Pair it with minPartitionBytes—if accumulated data hits the size threshold before the interval ends, it writes immediately; otherwise, it waits until the interval expires (your timeout fallback).

Example config snippet:

spark.conf.set("spark.sql.files.minPartitionBytes", "134217728") // 128MB
spark.conf.set("spark.sql.streaming.fileSink.rollingInterval", "300s") // 5-minute timeout

Spark doesn’t have a direct "min rows per file" config, but you can implement this with custom stateful logic, especially in Structured Streaming:

• Use foreachBatch to intercept each microbatch, then accumulate rows in memory (or a temporary view) until you hit your target row count.
• When the count is reached, write the batch to HDFS/Hive; if the microbatch ends without hitting the count, hold onto the data and combine it with the next batch.
• Add a timeout (tracked via a state variable) to avoid waiting forever if data trickles in slowly.

Simplified code example:

var accumulatedRows: DataFrame = null
val targetMinRows = 100000 // 100k rows per file
val timeoutMinutes = 5
var lastWriteTime = System.currentTimeMillis()

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
  accumulatedRows = if (accumulatedRows == null) batchDF else accumulatedRows.union(batchDF)
  
  // Check if we hit min rows or timeout
  val rowCount = accumulatedRows.count()
  val currentTime = System.currentTimeMillis()
  if (rowCount >= targetMinRows || (currentTime - lastWriteTime) > timeoutMinutes * 60 * 1000) {
    accumulatedRows.write.mode("append").saveAsTable("your_hive_table")
    accumulatedRows = null
    lastWriteTime = currentTime
  }
}.start()

Even with strict controls, you might get tiny files (e.g., from final batches when the stream stops). Clean these up with:

• Hive’s CONCATENATE command: For ORC/Parquet tables, run ALTER TABLE your_table PARTITION (dt='2024-05-20') CONCATENATE; to merge small files into larger ones.
• Spark’s OPTIMIZE command: If using Delta Lake, OPTIMIZE your_table ZORDER BY (key_column); merges small files and optimizes query performance.

• Balance size/row thresholds with latency needs—higher thresholds mean fewer files but longer wait times for data to be available in Hive.
• If you’re still using DStream, consider migrating to Structured Streaming—it has better built-in support for file sink controls and state management.
• Monitor your HDFS file system regularly to catch edge cases where tiny files slip through.

内容的提问来源于stack exchange，提问作者DigitalFailure