如何在Databricks中导出Delta表数据并写入TXT文件

以下是适配Databricks环境的完整PySpark代码，直接运行即可实现需求：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, row_number, concat_ws, monotonically_increasing_id
from pyspark.sql.types import StructType, ArrayType, MapType
from pyspark.sql.window import Window
import pandas as pd
from pyspark.sql.functions import pandas_udf

# 初始化SparkSession（Databricks环境可省略，保留以兼容本地调试）
spark = SparkSession.builder.appName("DeltaToTabSeparatedTxt").getOrCreate()

def flatten_complex_types(df):
    """转换复杂数据类型为可导出的字符串格式"""
    for field in df.schema.fields:
        # 处理Struct类型：拆分为单独列
        if isinstance(field.dataType, StructType):
            for sub_field in field.dataType.fields:
                df = df.withColumn(f"{field.name}_{sub_field.name}", col(f"{field.name}.{sub_field.name}"))
            df = df.drop(field.name)
        # 处理Array类型：转为逗号分隔字符串
        elif isinstance(field.dataType, ArrayType):
            df = df.withColumn(field.name, concat_ws(",", col(field.name)))
        # 处理Map类型：转为key:value格式的字符串
        elif isinstance(field.dataType, MapType):
            df = df.withColumn(field.name, concat_ws(",", concat_ws(":", col(field.name).keys(), col(field.name).values())))
    return df

# 1. 读取Delta表（替换为你的表路径或表名，如spark.table("your_db.your_table")）
delta_df = spark.read.format("delta").load("/path/to/your/delta/table")

# 2. 转换复杂数据类型
flattened_df = flatten_complex_types(delta_df)

# 3. 添加行号并划分数据块（每5万行一个块）
window_spec = Window.orderBy(monotonically_increasing_id())
df_with_row = flattened_df.withColumn("row_num", row_number().over(window_spec))
df_with_chunk = df_with_row.withColumn("chunk_id", (col("row_num") - 1) // 50000)

# 4. 定义导出函数，每个块生成一个带表头的TXT文件
@pandas_udf("void")
def export_chunk(chunk_df: pd.DataFrame) -> None:
    # 移除辅助列
    export_data = chunk_df.drop(columns=["row_num", "chunk_id"])
    # 获取当前块ID作为文件名后缀
    chunk_id = chunk_df["chunk_id"].iloc[0]
    # 替换为你的输出路径（DBFS路径需加/dbfs/前缀，如/dbfs/mnt/output/chunk_xxx.txt）
    output_file = f"/dbfs/your/output/directory/chunk_{chunk_id}.txt"
    # 导出为制表符分隔的TXT，包含表头
    export_data.to_csv(output_file, sep="\t", index=False, header=True)

# 5. 按块分组执行导出
df_with_chunk.groupBy("chunk_id").apply(export_chunk).collect()

关键说明 #

• 复杂类型处理：自动将Struct拆分为独立列、Array转为逗号分隔字符串、Map转为key:value格式，确保导出内容为纯文本。
• 分块逻辑：通过行号计算每5万行划分为一个文件，523万行将生成105个文件（含最后一个不足5万行的文件）。
• 路径替换：务必修改/path/to/your/delta/table为你的Delta表路径/表名，/dbfs/your/output/directory为你的输出目录。
• Databricks注意事项：输出路径需使用DBFS可访问路径，若使用分布式存储（如S3），需替换为对应路径格式。

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