Python/PySpark是否有替代SAS生成固定宽度文件的等效代码

原SAS代码核心逻辑为基于1起始的位置偏移指定每个字段的输出位置，字段不足长度自动补空格，超长自动截断，最终生成无分隔符的固定宽度文本文件，字段偏移对应关系如下：

• sequence_nbr：起始位1，长度9
• customer_name：起始位10，长度120
• append_year：起始位130，长度4
• append_month：起始位134，长度2
• zip_code：起始位136，长度5
• key_ele：起始位141，长度9
• confidence_code：起始位150，长度2
• address：起始位152，长度64
• city：起始位216，长度50
• state：起始位266，长度50
• grade：起始位316，长度11
• code：起始位327，长度可按业务需求自定义

方案1：Python原生实现（小数据量/pandas场景） #

逐行按偏移位置填充字符，对齐规则、截断逻辑和SAS默认行为保持一致，可直接替换对应SAS逻辑：

import pandas as pd

# 字段配置：(字段名, 1起始偏移, 字段长度, 对齐方式)
# 长度规则：当前字段起始位到下一个字段起始位的差值，最后一个字段按业务要求自定义
field_config = [
    ("sequence_nbr", 1, 9, "left"),
    ("customer_name", 10, 120, "left"),
    ("append_year", 130, 4, "left"),
    ("append_month", 134, 2, "left"),
    ("zip_code", 136, 5, "left"),
    ("key_ele", 141, 9, "left"),
    ("confidence_code", 150, 2, "left"),
    ("address", 152, 64, "left"),
    ("city", 216, 50, "left"),
    ("state", 266, 50, "left"),
    ("grade", 316, 11, "left"),
    ("code", 327, 10, "left")
]
total_width = 327 + 10 - 1  # 整行总宽度

def gen_fw_line(row):
    line_arr = [" "] * total_width
    for col, start, length, align in field_config:
        val = str(row[col]) if pd.notna(row[col]) else ""
        val = val[:length]  # 超长截断，和SAS默认行为一致
        # 对齐填充
        padded = val.ljust(length) if align == "left" else val.rjust(length)
        # 转0基索引写入
        start_idx = start - 1
        line_arr[start_idx:start_idx+length] = list(padded)
    return "".join(line_arr)

# 读取输入数据，替换为实际数据源读取逻辑
df = pd.read_parquet("input_trans_1.parquet")
# 生成固定宽度行并写入文件
lines = df.apply(gen_fw_line, axis=1)
with open("out.txt", "w", encoding="utf-8", newline="") as f:
    f.write("\n".join(lines))

方案2：PySpark实现（大数据分布式场景） #

用Spark内置字符串函数批量处理，避免逐行序列化开销，适配分布式计算场景：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit, lpad, rpad, substring, concat

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

# 字段配置和Python版本完全一致
field_config = [
    ("sequence_nbr", 1, 9, "left"),
    ("customer_name", 10, 120, "left"),
    ("append_year", 130, 4, "left"),
    ("append_month", 134, 2, "left"),
    ("zip_code", 136, 5, "left"),
    ("key_ele", 141, 9, "left"),
    ("confidence_code", 150, 2, "left"),
    ("address", 152, 64, "left"),
    ("city", 216, 50, "left"),
    ("state", 266, 50, "left"),
    ("grade", 316, 11, "left"),
    ("code", 327, 10, "left")
]

# 读取输入数据，空值统一填空字符串
df = spark.read.parquet("input_trans_1.parquet").fillna("")

processed_segments = []
prev_end = 0
for col_name, start, length, align in field_config:
    # 补字段前的空格，匹配偏移位置
    pre_pad_len = start - prev_end - 1
    if pre_pad_len > 0:
        processed_segments.append(lit(" " * pre_pad_len))
    # 字段值转字符串、截断、对齐填充
    val = col(col_name).cast("string")
    truncated = substring(val, 1, length)
    padded = rpad(truncated, length, " ") if align == "left" else lpad(truncated, length, " ")
    processed_segments.append(padded)
    prev_end = start + length - 1

# 拼接为整行写入
final_df = df.select(concat(*processed_segments).alias("line"))
final_df.write.text("out_fw_dir", mode="overwrite")

注意事项：

• SAS默认字符型字段左对齐、数值型字段右对齐，如果需要和SAS输出完全一致，把数值类字段的align参数改为right即可
• 编码可根据业务要求在写入时调整，支持GBK、UTF-8等常见编码
• 最后一个字段code的长度可按实际业务需求修改配置即可

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