PySpark写入大型机EBCDIC编码文本文件时 junk字符问题求助

核心思路 #

PySpark的text写入格式不支持自定义编码，且COMP3属于二进制数据，不应被解码为字符串处理。正确的做法是直接生成二进制字节流，按大型机要求的固定长度记录格式输出，绕开字符串编码转换的问题。

步骤1：修正数据处理逻辑，直接生成二进制字节 #

• 字符串类型字段：直接编码为cp037（EBCDIC）字节，按大型机要求的固定长度补位（通常用EBCDIC空格0x40补位）。
• 数值类型字段：通过自定义UDF生成COMP3格式的二进制字节（无需解码为字符串）。
• 拼接所有字段的字节，形成一条完整的记录字节数组（可根据需求添加EBCDIC换行符0x15）。

步骤2：用RDD二进制写入替代DataFrame text写入 #

利用PySpark RDD的saveAsBinaryFile方法直接写入二进制数据，避免编码转换问题：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("EBCDIC_COMP3_Output").getOrCreate()

# 读取Hive表
df = spark.sql("SELECT str_col, num_col FROM your_hive_table")

# 假设自定义UDF已注册为comp3_udf，输入数值返回COMP3格式的bytes
df = df.withColumn("comp3_num", comp3_udf(df.num_col))

def convert_row_to_binary(row):
    # 处理字符串字段：固定长度10，不足用EBCDIC空格补位
    str_ebcdic = row.str_col.ljust(10).encode("cp037")
    # 取出COMP3二进制数据
    comp3_data = row.comp3_num
    # 拼接记录，添加EBCDIC换行符（0x15），可根据大型机要求调整
    return str_ebcdic + comp3_data + b'\x15'

# 转换为二进制RDD并保存
binary_rdd = df.rdd.map(convert_row_to_binary)
# coalesce(1)确保输出单个文件，适合大型机读取
binary_rdd.coalesce(1).saveAsBinaryFile("/path/to/large_machine_input")

步骤3：验证输出文件 #

生成的二进制文件可通过以下方式验证：

• 使用xxd或二进制编辑器查看字节内容，确认字符串字段为cp037编码，COMP3字段符合预期格式。
• 发送至大型机环境测试读取，确认无垃圾字符或格式错误。

垃圾字符问题的根源 #

之前出现的X'78'等垃圾字符，是因为将COMP3二进制数据错误地解码为cp037字符串导致的。COMP3是压缩十进制的二进制存储格式，并非可打印的EBCDIC字符，直接解码会将二进制值映射为随机的EBCDIC字符，从而产生垃圾数据。直接保留二进制并输出即可解决此问题。

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