Azure Databricks新手求助：从BDL复制文件到ADLS并按条件拆分

一、BDL文件复制到ADLS #

如果BDL和ADLS已挂载到Databricks（或可通过原生ABFS路径直接访问），用dbutils.fs.cp命令即可完成单文件复制：

# 替换为实际的BDL源路径和ADLS目标路径
dbutils.fs.cp("dbfs:/mnt/bdl/your_source_file.csv", "dbfs:/mnt/adls/your_target_file.csv", recurse=False)

• 参数说明：
  • recurse=False：针对单个文件复制；若复制整个文件夹，改为recurse=True
  • 路径支持原生ABFS格式，比如abfss://container@your-bdl-storage.dfs.core.windows.net/source.csv

二、按指定条件拆分文件为5份 #

根据需求分两种常见场景实现：

场景1：按行数平均拆分 #

如果需要将文件平均分成5份，用randomSplit实现：

# 读取ADLS中的目标文件
df = spark.read.csv("dbfs:/mnt/adls/your_target_file.csv", header=True, inferSchema=True)

# 按均等权重拆分为5个DataFrame（seed保证拆分结果可复现）
df_splits = df.randomSplit([0.2, 0.2, 0.2, 0.2, 0.2], seed=42)

# 逐个写入ADLS指定目录
for idx, split_df in enumerate(df_splits):
    # 若要生成单个文件，添加coalesce(1)；否则会生成多个分区文件
    split_df.coalesce(1).write.csv(
        f"dbfs:/mnt/adls/split_files/split_{idx+1}",
        header=True,
        mode="overwrite"
    )
    
    # 可选：将生成的part文件重命名为指定文件名
    source_file = dbutils.fs.ls(f"dbfs:/mnt/adls/split_files/split_{idx+1}")[0].path
    dbutils.fs.mv(source_file, f"dbfs:/mnt/adls/split_files/split_{idx+1}.csv")
    # 删除空的拆分目录
    dbutils.fs.rm(f"dbfs:/mnt/adls/split_files/split_{idx+1}", recurse=True)

场景2：按业务条件拆分 #

如果需要按特定字段（如地区、用户类型）拆分，用filter实现：

# 读取ADLS中的目标文件
df = spark.read.csv("dbfs:/mnt/adls/your_target_file.csv", header=True, inferSchema=True)

# 定义5个拆分条件（替换为实际业务规则）
split_rules = [
    "customer_type = 'Premium'",
    "customer_type = 'Standard'",
    "customer_type = 'Basic'",
    "age >= 30 AND age < 45",
    "age >= 45"
]

# 循环处理每个条件并写入文件
for idx, rule in enumerate(split_rules):
    filtered_df = df.filter(rule)
    filtered_df.coalesce(1).write.csv(
        f"dbfs:/mnt/adls/split_files/condition_{idx+1}",
        header=True,
        mode="overwrite"
    )
    
    # 可选：重命名文件
    source_file = dbutils.fs.ls(f"dbfs:/mnt/adls/split_files/condition_{idx+1}")[0].path
    dbutils.fs.mv(source_file, f"dbfs:/mnt/adls/split_files/condition_{idx+1}.csv")
    dbutils.fs.rm(f"dbfs:/mnt/adls/split_files/condition_{idx+1}", recurse=True)

注意事项 #

• 若未挂载存储，可通过dbutils.fs.mount挂载ADLS Gen2示例：

  dbutils.fs.mount(
      source="abfss://your-container@your-storage-account.dfs.core.windows.net/",
      mount_point="/mnt/adls",
      extra_configs={"fs.azure.account.key.your-storage-account.dfs.core.windows.net": "your-storage-key"}
  )
  

• 支持Parquet、JSON等其他格式，只需将read.csv/write.csv替换为对应格式方法（如read.parquet）
• coalesce(1)用于合并为单个文件，若数据量极大，建议保留多分区文件以提升性能

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