PySpark中多XSLT并行转换的实现方案咨询

问题背景 #

• 需求：在Databricks Notebook中处理大量复杂XML文件，每个文件需通过XSLT 2.0转换，输出对应多张表
• 当前瓶颈：
  • 使用PySaxon串行处理30-40个文件效率极低
  • 尝试将转换逻辑放入UDF，导致集群Python内核无响应（推测为每行重复初始化PySaxon处理器，内存占用过高）
• 已尝试方案：通过ADF调用Notebook单文件处理，每个文件对应一个Notebook实例，成本开销过大

现有串行处理代码 #

xlst = spark.read.text(file_path_xslt, wholetext=True)  # 加载完整XSLT文本
stringxslt = xlst.head()[0]
listOfTransformedXML = []
with PySaxonProcessor(license=False) as proc:
    xsltproc = proc.new_xslt_processor()
    xsltproc.compile_stylesheet(stylesheet_text=stringxslt)
    for file in dbutils.fs.ls(pathToADLSStorage):
        data = spark.read.text(file[0])
        stringxml = data.head()[0]
        document = proc.parse_xml(xml_text=stringxml)
        xsltproc.set_source(xdm_node=document)
        output2 = xsltproc.transform_to_string()
        # 移除XML声明，避免解析报错
        stringforxmlobj = output2.replace("<?xml version=\"1.0\" encoding=\"UTF-8\"?>","")
        listOfTransformedXML.append(stringforxmlobj)

多线程尝试的问题 #

参考示例改造为多线程处理后，实际仍为串行执行（处理3个文件的耗时等于单独处理总和），集群仅在1个执行器上创建2个任务，推测线程仅在主节点运行，未实现分布式处理。

可行并行转换方案 #

方案1：PySpark RDD分区级分布式处理（纯Python） #

核心思路：将XSLT文本广播到所有执行器，每个分区内仅初始化一次PySaxon处理器，复用编译后的样式表，避免重复开销。

# 1. 读取XSLT样式表文本
xslt_text = spark.read.text(file_path_xslt, wholetext=True).head()[0]

# 2. 获取所有XML文件路径，构建DataFrame
xml_file_paths = [f.path for f in dbutils.fs.ls(pathToADLSStorage)]
xml_df = spark.createDataFrame([(path,) for path in xml_file_paths], ["file_path"])

# 3. 广播XSLT文本到所有执行器
broadcast_xslt = spark.sparkContext.broadcast(xslt_text)

# 4. 定义分区处理函数：每个分区初始化一次处理器
def process_partition(partition):
    from saxonche import PySaxonProcessor
    
    # 分区内复用处理器和编译后的样式表
    with PySaxonProcessor(license=False) as proc:
        xslt_proc = proc.new_xslt_processor()
        xslt_proc.compile_stylesheet(stylesheet_text=broadcast_xslt.value)
        
        for row in partition:
            file_path = row.file_path
            # 读取单个XML文件完整内容
            xml_content = spark.read.text(file_path, wholetext=True).head()[0]
            # 解析XML并转换
            doc = proc.parse_xml(xml_text=xml_content)
            transformed_str = xslt_proc.transform_to_string(xdm_node=doc)
            # 清理XML声明
            cleaned_str = transformed_str.replace('<?xml version="1.0" encoding="UTF-8"?>', '')
            yield cleaned_str

# 5. 执行分布式转换并收集结果
transformed_rdd = xml_df.rdd.mapPartitions(process_partition)
result_list = transformed_rdd.collect()

方案2：使用Saxon Java API（稳定性最优） #

PySaxon在Python分布式环境下的序列化和资源管理不如Java版Saxon成熟，直接使用Java API结合PySpark，可避免Python内存溢出问题。

步骤1：在Scala Cell中定义转换逻辑并注册UDF

import net.sf.saxon.s9api._
import java.io.{ByteArrayInputStream, StringWriter}

// 加载并编译XSLT样式表
val xsltText = spark.read.textFile(file_path_xslt).collect().mkString("\n")
val processor = new Processor(false)
val xsltCompiler = processor.newXsltCompiler()
val xsltExecutable = xsltCompiler.compile(new StreamSource(new ByteArrayInputStream(xsltText.getBytes("UTF-8"))))

// 注册分布式UDF
spark.udf.register("transformXml", (xmlContent: String) => {
  val docBuilder = processor.newDocumentBuilder()
  val xmlDoc = docBuilder.build(new StreamSource(new ByteArrayInputStream(xmlContent.getBytes("UTF-8"))))
  val xsltTransformer = xsltExecutable.load()
  xsltTransformer.setSource(xmlDoc)
  
  val resultWriter = new StringWriter()
  xsltTransformer.setDestination(new StreamResult(resultWriter))
  xsltTransformer.transform()
  
  // 移除XML声明
  resultWriter.toString.replace("<?xml version=\"1.0\" encoding=\"UTF-8\"?>", "")
})

步骤2：在Python Cell中调用UDF处理XML

# 读取所有XML文件内容
xml_df = spark.read.text(pathToADLSStorage, wholetext=True).withColumnRenamed("value", "xml_content")

# 应用UDF完成转换
transformed_df = xml_df.selectExpr("transformXml(xml_content) as transformed_xml")

# 收集结果或继续后续表解析逻辑
result_list = transformed_df.select("transformed_xml").rdd.flatMap(lambda x: [x[0]]).collect()

方案3：优化Python UDF（最小改动） #

若坚持使用Python UDF，通过延迟初始化实现每个执行器仅创建一次PySaxon处理器，避免每行重复初始化的内存开销。

from saxonche import PySaxonProcessor
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 全局变量：每个执行器仅初始化一次
processor = None
xslt_proc = None

def init_processor(xslt_text):
    global processor, xslt_proc
    if processor is None:
        processor = PySaxonProcessor(license=False)
        xslt_proc = processor.new_xslt_processor()
        xslt_proc.compile_stylesheet(stylesheet_text=xslt_text)

# 定义UDF
@udf(StringType())
def transform_xml_udf(xml_content):
    init_processor(broadcast_xslt.value)
    doc = processor.parse_xml(xml_text=xml_content)
    transformed_str = xslt_proc.transform_to_string(xdm_node=doc)
    return transformed_str.replace('<?xml version="1.0" encoding="UTF-8"?>', '')

# 广播XSLT文本
xslt_text = spark.read.text(file_path_xslt, wholetext=True).head()[0]
broadcast_xslt = spark.sparkContext.broadcast(xslt_text)

# 读取XML并执行转换
xml_df = spark.read.text(pathToADLSStorage, wholetext=True)
transformed_df = xml_df.withColumn("transformed", transform_xml_udf(xml_df.value))
result_list = transformed_df.select("transformed").rdd.flatMap(lambda x: [x[0]]).collect()

方案对比 #

• 方案1：纯Python实现，无需额外依赖，分区级复用处理器，内存占用可控，适合中小规模数据
• 方案2：Java版Saxon稳定性更高，分布式支持更好，适合大规模复杂XML处理
• 方案3：代码改动最小，适合快速迭代，但需注意执行器内存配置，避免内存溢出

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