Databricks Delta Lake流读取AWS CSV文件问题求助

先拆解下你遇到的核心问题：

• 读取单个文件：你用的data_from_****.csv通配符可能只匹配到了单个2018年的文件，应该直接指向文件夹路径，Spark会自动扫描文件夹内所有CSV
• 无法获取年份：skiprows=3直接跳过了包含年份的第1行，得先读取该行提取年份后，再过滤无效行
• 流处理逻辑：需要确保能批量处理现有文件或监听文件夹内新增文件

方案一：从文件第1行提取年份（匹配你的需求） #

这种方式需要先读取所有行，提取每个文件第1行的年份后，再过滤掉前3行无效内容：

1. 修改路径为文件夹路径，不再指定单个文件：

path_folder = "/mnt/path/path_data/path_data_year/"

2. 读取所有文件内容，同时记录每行所属的文件名：

from pyspark.sql.functions import input_file_name, first, row_number, lit
from pyspark.sql.window import Window

df_raw = spark.readStream \
    .format("csv") \
    .option("delimiter", ";") \
    .option("encoding", "iso-8859-1") \
    .option("header", "false")  # 先不设置表头，因为前3行不是有效数据表头
    .load(path_folder) \
    .withColumn("file_path", input_file_name())

3. 提取每个文件的年份（第1行第B列对应CSV的_c1列，因CSV列索引从0开始）：

# 按文件分组，取每个文件第1行的_c1值作为年份
df_file_year = df_raw.groupBy("file_path") \
    .agg(first(col("_c1")).alias("year"))

# 给原始数据关联年份，再过滤掉前3行
window_spec = Window.partitionBy("file_path").orderBy(lit(1))
df_filtered = df_raw.join(df_file_year, on="file_path", how="inner") \
    .withColumn("row_num", row_number().over(window_spec)) \
    .filter(col("row_num") > 3) \
    .drop("row_num", "file_path")

4. 应用预定义Schema，给列重命名：

# 将_c0、_c1等默认列名替换为你schema中的实际列名
df_final = df_filtered.select(
    col("_c0").alias("列名1"),
    col("_c1").alias("列名2"),
    # ... 依次对应其他列
    col("year").alias("year")
)

5. 写入Delta Lake：

df_final.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/mnt/path/checkpoints/despesUF_funcao")  # 必须设置检查点路径
    .table("bronze.despesUF_funcao")

方案二：从文件名提取年份（更简单高效） #

如果你的文件名data_from_YYYY.csv格式绝对规范，直接从文件名提取4位年份更省事，无需处理前3行的年份信息：

from pyspark.sql.functions import regexp_extract

path_folder = "/mnt/path/path_data/path_data_year/"

df_bronze_despesUF_funcao = spark.readStream \
    .format("csv") \
    .schema(schema) \
    .options(
        header='true', 
        delimiter=';', 
        encoding='iso-8859-1', 
        skiprows=3
    ) \
    .load(path_folder) \
    .withColumn("year", regexp_extract(input_file_name(), r"data_from_(\d{4})\.csv", 1))

之后直接用和方案一相同的写入代码，将数据存入Delta表即可。

关键注意事项 #

• 流处理默认监听文件夹内新增文件，若要一次性处理现有所有文件，可添加.option("maxFilesPerTrigger", 10)控制每次触发处理的文件数量
• 确保预定义的schema与过滤后的数据列完全匹配，避免类型错误
• 检查所有文件的编码、分隔符是否统一，防止读取出现乱码或列错位

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