PySpark列级血缘实现:获取两DataFrame源列到目标列的映射方法
Spark DataFrame列映射生成及导出方案
以下是可直接运行的完整实现流程,覆盖列映射定义、目标DF生成、映射关系文件导出三个核心环节:
1. 定义列映射规则
用字典存储目标列到源列的对应关系,可根据你的业务需求灵活调整:
# 键为目标DF列名,值为源DF列名,对应你给出的期望输出规则 col_mapping = { "col1": "col1", "col2": "col2", "col3": "col2" }
2. 按映射规则生成目标DataFrame
无需逐行写withColumn,批量生成目标DF更高效:
from pyspark.sql.functions import col # 读取源数据 source_df = spark.read.parquet('/temp/file.parquet') # 批量生成目标DF target_df = source_df.select( *[col(source_col).alias(target_col) for target_col, source_col in col_mapping.items()] ) # 验证生成的目标DF数据 target_df.show()
3. 导出映射关系为文本文件
方案A:导出制表符分隔的结构化文件
适合后续程序读取解析:
# 把映射规则转为Spark DataFrame mapping_data = [(target_col, source_col) for target_col, source_col in col_mapping.items()] mapping_df = spark.createDataFrame(mapping_data, schema=["Target", "Source"]) # 输出为单个文本文件,coalesce(1)用于避免文件分片 mapping_df.coalesce(1).write \ .option("header", "true") \ .option("delimiter", "\t") \ .mode("overwrite") \ .csv("/你要保存的输出路径/column_mapping")
输出文件内容和你给出的期望格式完全一致:
Target Source col1 --> col1 col2 --> col2 col3 --> col2
方案B:导出带-->标识的纯文本
适合人工阅读:
from pyspark.sql.functions import concat, lit # 拼接成要求的格式 mapping_df = mapping_df.select( concat(col("Target"), lit(" --> "), col("Source")).alias("mapping_content") ) # 输出纯文本,不带表头 mapping_df.coalesce(1).write \ .option("header", "false") \ .mode("overwrite") \ .text("/你要保存的输出路径/column_mapping")
内容的提问来源于stack exchange,提问作者Kannan Radhakrishnan
相关产品推荐
相关产品推荐

