在Azure Databricks中高效读取大量单记录JSON文件的最优方案

针对你这种15万+单记录小JSON文件的读取场景，确实会因为文件数量过多导致Spark的元数据扫描开销极大，以下几个方法能帮你显著提速，同时还能保留每条记录的文件来源：

1. 优化文件路径与分区推断 #

• 如果你的目录结构本来就是按year/month/day/hour分层的，建议将路径调整为分区格式（比如src_folder/year=*/month=*/day=*/hour=*/*/*.json），让Spark自动识别分区列，减少不必要的目录扫描。Spark在读取分区目录时，会跳过对非分区目录的元数据遍历，大幅降低初始化时间。
• 尽量避免多层通配符*，如果知道小时下的子目录命名规则，直接指定具体范围，减少Spark需要遍历的目录数量。

2. 调整Spark小文件读取配置 #

在读取数据前，设置以下Spark配置，优化小文件的处理逻辑：

// 增大每个分区的字节数，减少任务数量
spark.conf.set("spark.sql.files.maxPartitionBytes", "256m")
// 提高打开文件的成本权重，让Spark优先合并更多小文件到一个分区
spark.conf.set("spark.sql.files.openCostInBytes", "64m")
// 忽略损坏或缺失的文件，避免流程中断
spark.conf.set("spark.sql.json.ignoreCorruptFiles", "true")
spark.conf.set("spark.sql.json.ignoreMissingFiles", "true")

这些配置能有效降低Spark调度和文件IO的开销，尤其针对海量小文件场景。

3. 使用WholeText模式读取后解析 #

由于每个文件仅包含一条JSON记录，推荐先用wholetext模式批量读取文件内容，再统一解析JSON，同时保留文件名：

import org.apache.spark.sql.functions.{from_json, input_file_name}

val schema_variable = <schema>
val file_path = "src_folder/year/month/day/hour/*/*.json"

val df = spark.read
  .option("wholetext", true)
  .text(file_path)
  .withColumn("file_name", input_file_name())
  .withColumn("json_data", from_json($"value", schema_variable))
  .select("json_data.*", "file_name")

这种方式跳过了Spark读取阶段逐文件的JSON schema验证，将解析逻辑放到后续的转换阶段，并行度更高，能显著缩短读取时间。

4. 预处理合并小文件（非实时场景可选） #

如果你的数据是非实时的，可以定期用Azure工具（比如Azure Data Factory、AzCopy或Databricks的dbutils.fs）将小文件合并为大文件，同时把原文件名作为字段写入合并后的记录中。例如合并时，将每个小文件的内容加上file_name字段后再合并，这样既解决了小文件问题，又不会丢失谱系信息。

5. 集群资源优化 #

• 选择配置更高的Worker节点（比如Standard_DS4_v2及以上），增加Worker数量，确保有足够的CPU和内存处理元数据扫描和文件IO。
• 调整Executor参数：增大spark.executor.cores（比如设置为4）和spark.executor.memory（比如设置为16G），让每个Executor能处理更多的文件任务。
• 开启文件系统缓存：spark.conf.set("spark.databricks.io.cache.enabled", "true")，减少重复读取的开销。

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