Azure Databricks如何从挂载点动态读取新增文件并追加至Delta表？

现有环境与已执行操作 #

• 已创建挂载点：dbutils.fs.ls("/mnt/mount/raw2")
• 初始文件customer1.csv内容：

| ID| Name| Place|
+---+------+------+
|101| Hari| Tcr|
|102| John| Bgr|

• 已完成的读取与Delta表写入：

# 读取初始CSV文件到DataFrame
df = spark.read.option("header", "true").csv("/mnt/mount/raw2")

# 将数据写入Delta表（覆盖模式）
df.write.format("delta").mode("overwrite").save("/mnt/mount/raw2/customer_data")

新增文件情况 #

挂载点新增customer2.csv，内容如下（无表头）：

103|Stefen| Hyd|
|104| Devid| Bgr|
|105| Wager|London|

实现动态读取新增文件并追加的两种方案 #

方案一：手动跟踪已处理文件（适合一次性增量处理） #

通过维护已处理文件列表，筛选未处理的新增文件，读取后追加至Delta表：

from pyspark.sql.functions import regexp_replace, col

# 1. 获取挂载点下所有CSV文件路径
all_csv_files = [file.path for file in dbutils.fs.ls("/mnt/mount/raw2") if file.path.endswith(".csv")]

# 2. 读取已处理文件记录（首次执行时文件不存在则初始化为空列表）
try:
    processed_files = spark.read.text("/mnt/mount/raw2/processed_files.txt").rdd.map(lambda row: row[0]).collect()
except:
    processed_files = []

# 3. 筛选出未处理的新增文件
new_files = [file for file in all_csv_files if file not in processed_files]

if new_files:
    # 4. 读取新增文件（注意customer2.csv无表头，需指定header=false并手动添加列名）
    df_new = spark.read.option("header", "false").option("delimiter", "|").csv(new_files)
    df_new = df_new.toDF("ID", "Name", "Place")
    
    # 5. 清理数据：去除字段首尾的竖线、过滤空值
    df_clean = df_new.withColumn("ID", regexp_replace(col("ID"), "^\\|?", "")) \
                     .withColumn("Place", regexp_replace(col("Place"), "\\|?$", "")) \
                     .filter(col("ID").isNotNull())
    
    # 6. 追加数据到Delta表
    df_clean.write.format("delta").mode("append").save("/mnt/mount/raw2/customer_data")
    
    # 7. 更新已处理文件记录
    with open("/dbfs/mnt/mount/raw2/processed_files.txt", "a") as f:
        for file in new_files:
            f.write(file + "\n")

方案二：使用Delta Lake Auto Loader（推荐，适合持续增量场景） #

Auto Loader会自动检测挂载点新增的文件，无需手动跟踪，还支持schema自动推断与演进：

# 启动流处理，自动读取新增CSV文件并追加至Delta表
spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "csv") \
    .option("cloudFiles.schemaLocation", "/mnt/mount/raw2/schema_store")  # 存储自动推断的schema
    .option("header", "true")  # 适配有表头的文件，无表头文件会自动处理
    .load("/mnt/mount/raw2") \
    .writeStream \
    .format("delta") \
    .option("checkpointLocation", "/mnt/mount/raw2/checkpoint")  # 记录流处理进度，避免重复处理
    .start("/mnt/mount/raw2/customer_data")

注意事项 #

• 确保新增文件与原表数据格式一致，若存在无表头、分隔符差异等情况，需在读取时统一处理
• 手动跟踪方案需注意文件路径的准确性，避免重复或遗漏处理
• Auto Loader适合持续产生增量文件的场景，流处理会持续运行并自动处理新增数据

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