Azure Databricks中cat合并大文件报错的根因及解决方法咨询

问题描述 #

在Azure Databricks Notebook中使用%sh执行cat命令合并文件时，data_files文件夹下有约1200个CSV文件，总大小约300GB。运行以下脚本时，有时能正常完成，但有时会抛出错误cat: write error: Resource temporarily unavailable，且生成的output.txt文件无数据：

err=$(cat /dbfs/mnt/devl/header_file/*.csv /dbfs/mnt/devl/data_files/*.csv 2>&1 > /dbfs/mnt/devl/output.txt)
RC=$?
if [ $RC -ne 0 ]; then
  echo "Error code : $RC"
  echo "Error msg : $err"
fi

根本原因 #

1. DBFS分布式IO临时资源竞争：DBFS是分布式文件系统，直接通过shell命令写入时，可能遇到集群节点间的IO资源争抢、临时限流或存储节点负载波动，导致写入操作暂时无法获取资源。
2. shell参数数量超限：1200个文件通过通配符展开后，可能超过系统的ARG_MAX限制（单个命令能接收的最大参数数量），此时系统无法处理过多参数，触发写入失败。
3. 单进程资源瓶颈：单个cat进程处理300GB数据合并，会占用大量内存和IO带宽，当节点资源不足时，会触发临时资源不可用的错误。

解决方法 #

方法1：循环逐个追加文件，分散压力 #

避免一次性传入大量文件参数，通过循环逐个追加，降低单次IO压力和参数数量：

# 先写入表头文件
cat /dbfs/mnt/devl/header_file/*.csv > /dbfs/mnt/devl/output.txt
# 遍历数据文件逐个追加
for file in /dbfs/mnt/devl/data_files/*.csv; do
  cat "$file" >> /dbfs/mnt/devl/output.txt
done

方法2：使用Spark原生API合并（推荐） #

Databricks基于Spark构建，用分布式计算处理大文件合并更稳定，避免单点瓶颈：

from pyspark.sql import SparkSession

# 初始化Spark会话
spark = SparkSession.builder.appName("MergeLargeCSV").getOrCreate()

# 读取所有CSV文件（包含表头）
df = spark.read.csv(
  ["dbfs:/mnt/devl/header_file/*.csv", "dbfs:/mnt/devl/data_files/*.csv"],
  header=True,
  inferSchema=False
)

# 写入临时目录（repartition(1)生成单个文件，可根据实际需求调整分区数）
df.repartition(1).write.mode("overwrite").option("header", "true").csv("dbfs:/mnt/devl/temp_merge")

# 找到生成的part文件并重命名为output.txt
part_files = [f.path for f in dbutils.fs.ls("dbfs:/mnt/devl/temp_merge") if f.path.endswith(".csv")]
if part_files:
  dbutils.fs.mv(part_files[0], "dbfs:/mnt/devl/output.txt")
  # 删除临时目录
  dbutils.fs.rm("dbfs:/mnt/devl/temp_merge", recurse=True)
else:
  print("No part file generated")

方法3：增加重试机制，应对临时错误 #

如果必须使用shell命令，添加重试逻辑来处理临时资源不可用的情况：

MAX_RETRIES=3
RETRY_COUNT=0
SUCCESS=0

while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do
  # 先删除可能存在的空文件
  rm -f /dbfs/mnt/devl/output.txt
  # 执行合并
  cat /dbfs/mnt/devl/header_file/*.csv /dbfs/mnt/devl/data_files/*.csv > /dbfs/mnt/devl/output.txt
  RC=$?
  if [ $RC -eq 0 ]; then
    SUCCESS=1
    echo "Merge completed successfully"
    break
  fi
  RETRY_COUNT=$((RETRY_COUNT+1))
  echo "Retry $RETRY_COUNT/$MAX_RETRIES failed, error code $RC"
  sleep 5  # 等待5秒后重试
done

if [ $SUCCESS -eq 0 ]; then
  echo "All $MAX_RETRIES retries failed"
fi

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