将Python类转换为Spark Delta行的适配问题求助

我正尝试改造现有Python包以适配Spark Structured Streaming，该包包含元数据二进制文件解析、光谱傅里叶变换等复杂子步骤。此前中间结果与最终结果通过SQLAlchemy存储于SQL数据库，现需转为Delta格式。

已通过静态定义列类型的UDF实现二进制文件解析部分：

fileparser = F.udf(File()._parseBytes, FileDelta.getSchema())

其中_parseBytes()方法接收二进制流并输出变量字典。

尝试以类似方式实现光谱生成：

spectrumparser = F.udf(lambda inputDict : vars(Spectrum(inputDict)), SpectrumDelta.getSchema())

但Spectrum()初始化方法会生成多个Pandas DataFrame作为字段，Executor节点执行时出现错误：

expected zero arguments for construction of ClassDict (for pandas.core.indexes.base._new_Index).  
This happens when an unsupported/unregistered class is being unpickled that requires construction arguments.  
Fix it by registering a custom IObjectConstructor for this class.

目前适配Delta格式花费过多精力，想知道是否有简便可行的解决方法？了解到可切换至Pandas on Spark API，但这似乎需要修改包内方法，是否需要重写整个包与解析器以原生适配PySpark？尝试用最小示例复现问题，但因包代码过于复杂难以实现。

• 避免在普通UDF中返回Pandas对象：Spark普通UDF无法正确序列化Pandas DataFrame这类复杂对象。修改光谱生成的UDF逻辑，将Spectrum实例中的Pandas DataFrame转换为Spark可序列化的原生类型（如字典列表），同步调整SpectrumDelta.getSchema()的结构（用ArrayType(StructType(...))匹配转换后的数据格式）：

  def process_spectrum(input_dict):
      spec = Spectrum(input_dict)
      spec_vars = vars(spec)
      # 将Pandas DataFrame转为字典列表
      for key, value in spec_vars.items():
          if isinstance(value, pd.DataFrame):
              spec_vars[key] = value.to_dict('records')
      return spec_vars
  
  spectrumparser = F.udf(process_spectrum, SpectrumDelta.getAdjustedSchema())
  

  这里getAdjustedSchema()需要把原Schema中对应DataFrame的字段替换为适配字典列表的结构。

• 用Pandas UDF简化适配，无需重写整个包：使用PySpark的Pandas UDF（Scalar类型）批量处理数据，Spark会自动处理Pandas与Spark类型的转换，无需修改包内核心逻辑：

  from pyspark.sql.functions import pandas_udf
  import pandas as pd
  
  @pandas_udf(SpectrumDelta.getAdjustedSchema())
  def spectrum_parser_pandas_udf(input_dicts: pd.Series) -> pd.Series:
      def process_single(d):
          spec = Spectrum(d)
          spec_vars = vars(spec)
          # 替换所有DataFrame为字典列表
          for k, v in spec_vars.items():
              if isinstance(v, pd.DataFrame):
                  spec_vars[k] = v.to_dict('records')
          return spec_vars
      return input_dicts.apply(process_single)
  

  这种方式比普通UDF效率更高，且能规避大部分跨节点序列化问题。

• 拆分复杂逻辑，分步处理：把光谱生成拆分为多个独立的Spark操作：先解析生成基础元数据并写入临时Delta表，再基于临时表数据单独处理傅里叶变换生成光谱数据，最后合并结果写入目标Delta表。每个步骤仅处理简单数据类型，减少复杂对象的跨节点传递。

• 不建议自定义序列化器：错误提示中提到的注册IObjectConstructor操作复杂，且在Spark分布式环境中稳定性差，生产环境不推荐使用。

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