PySpark中RDD pipe操作将Row转为字符串后如何转回Row对象

通过PySpark的pipe方法将RDD输出到基于stdin/stdout标准输入输出通信的外部进程，实现代码如下：

piped_rdd = rdd.pipe(exe_path)

检查返回的PipeLineRDD时发现，所有元素都被转换为字符串格式，数据示例：

["Row(ID='x123223=', FirstName='L', LastName='S')", "Row(ID='43454'.....)"]

需要将这类字符串还原为标准PySpark Row对象。

• 可信输入场景：直接用eval解析
  PySpark Row的默认字符串输出是合法Python表达式，若外部进程输出完全可信、不存在恶意注入风险，可直接通过eval完成转换，传入命名空间限定仅加载Row类降低风险：

  from pyspark.sql import Row
  row_rdd = piped_rdd.map(lambda line: eval(line, {"Row": Row}))
  

  警告：eval会执行传入字符串里的任意Python代码，禁止在输入来源不可控的场景使用，避免触发代码注入漏洞。

• 不可信输入场景：自定义安全解析逻辑
  若无法保证输入安全性，可通过正则匹配、字符串切割的方式提取Row字符串内的字段与值，手动构造Row对象，完全规避代码执行风险。针对示例的Row(字段='值')格式，参考实现：

  from pyspark.sql import Row
  import re
  
  def str_to_row(line: str) -> Row:
      # 去除首尾空白后剥离外层的Row(和末尾的)
      core_content = line.strip()[4:-1]
      # 匹配key='value'格式的键值对
      kv_pairs = re.findall(r"(\w+)='([^']*)'", core_content)
      return Row(**dict(kv_pairs))
  
  row_rdd = piped_rdd.map(str_to_row)
  

  若字段存在非字符串类型、嵌套结构，可对应调整正则规则补充类型转换逻辑。

• 源头优化方案（最推荐）
  直接依赖Row的默认字符串序列化稳定性差，不同PySpark版本可能调整Row的字符串输出格式，提升后续维护成本。建议在调用pipe前先将Row序列化为标准JSON格式，要求外部进程处理后也输出标准JSON，再统一反序列化为Row，兼容性和安全性都更优：

  import json
  # pipe前将Row转为JSON字符串
  json_input_rdd = rdd.map(lambda row: json.dumps(row.asDict()))
  # 调用外部进程
  piped_output_rdd = json_input_rdd.pipe(exe_path)
  # 解析JSON转回Row对象
  row_rdd = piped_output_rdd.map(lambda line: Row(**json.loads(line)))
  

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