Spark SaveAsTable写入Hive每次生成新Parquet文件的合并方案咨询

Alright, let's tackle this problem you're facing with small Parquet files piling up when appending JSON data to your Hive table. I've dealt with similar issues in production environments, so here are some practical, actionable solutions:

1. Control Output File Size During Write #

You can configure Spark to limit the number of records per output file or adjust parallelism to generate fewer, larger files. This prevents tiny files from being created every time you append data.

Add these settings either via SQL configs or directly in your DataFrameWriter:

// Option 1: Set via HiveContext configs
hiveContext.sql("SET spark.sql.files.maxRecordsPerFile=100000") // Adjust based on your record size (e.g., 100k records per ~128MB file)
hiveContext.sql("SET spark.sql.parquet.compression.codec=snappy") // Compression reduces file count and size

// Option 2: Set directly in the write operation
comment.write
  .option("maxRecordsPerFile", 100000)
  .mode("append")
  .saveAsTable("<your-table-name>")

Tweak the maxRecordsPerFile value based on your average record size—aim for files around the HDFS block size (usually 128MB or 256MB) for optimal performance.

2. Repartition New Data Before Appending #

If your incoming JSON data is split into many small partitions, you can repartition it to reduce the number of output files. Calculate a reasonable partition count based on the total size of your new data:

// Example: Repartition to 8 partitions for ~1GB of data (adjust based on your data size)
comment.repartition(8)
  .write
  .mode("append")
  .saveAsTable("<your-table-name>")

Avoid setting the partition count too low (risk of OOM) or too high (still gets small files)—test with your data to find the sweet spot.

3. Merge Existing Small Files #

If you already have a bunch of small files in your Hive table, you can merge them using Hive's built-in CONCATENATE command, which works specifically for Parquet tables:

// Merge small files in the table (works per partition if your table is partitioned)
hiveContext.sql("ALTER TABLE <your-table-name> CONCATENATE")

You can run this periodically (e.g., daily after your append job) to keep the table's file structure clean.

4. Use INSERT INTO Instead of saveAsTable (Optional) #

You mentioned relying on saveAsTable because of newline/carriage return characters, but INSERT INTO handles these special characters just as well. This approach gives you the same append functionality while letting you use SQL-based controls:

// Register your incoming data as a temporary view
comment.createOrReplaceTempView("temp_new_comments")

// Append to the Hive table
hiveContext.sql("""
  INSERT INTO TABLE <your-table-name>
  SELECT * FROM temp_new_comments
""")

Combine this with the maxRecordsPerFile config above to control file size during the insert.

Quick Note on Special Characters #

Just to clarify: Hive supports storing data with newlines and carriage returns—you only run into issues with legacy methods like LOAD DATA. Both saveAsTable and INSERT INTO via Spark handle these characters correctly, so you don't have to stick to one method exclusively.

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