如何检测XSD中的直接嵌套字段并正确转换为Spark Schema？

我正在实现一个将任意XSD Schema转换为PySpark可用Schema的函数，递归处理XSD元素时遇到瓶颈：

• 使用XPath ./{http://www.w3.org/2001/XMLSchema}element 会跳过嵌套在complexType里的元素（比如测试用例中Authors下的Author）
• 使用.//{http://www.w3.org/2001/XMLSchema}element 会把嵌套元素重复提取到顶层，导致结构混乱

需要找到正确的处理逻辑，既能识别任意深度的嵌套字段，又不会出现元素重复或遗漏。

当前实现代码 #

from lxml import etree
from pyspark.sql.types import StructType, StructField, StringType, ArrayType

def xsd_to_spark_schema(xsd_path):
    tree = etree.parse(xsd_path)
    root = tree.getroot()
    ns = {"xs": "http://www.w3.org/2001/XMLSchema"}
    
    def parse_element(element):
        fields = []
        # 问题出在XPath选择器
        child_elements = element.xpath(".//xs:element", namespaces=ns)
        # 或者尝试过 child_elements = element.xpath("./xs:element", namespaces=ns)
        for child in child_elements:
            name = child.get("name")
            type_name = child.get("type")
            if type_name.startswith("xs:"):
                spark_type = StringType()  # 简化处理，实际需映射更多类型
            else:
                complex_type = root.xpath(f"//xs:complexType[@name='{type_name}']", namespaces=ns)[0]
                spark_type = StructType(parse_element(complexType))
            
            if child.get("maxOccurs") == "unbounded":
                fields.append(StructField(name, ArrayType(spark_type), nullable=True))
            else:
                fields.append(StructField(name, spark_type, nullable=True))
        return fields
    
    root_elements = root.xpath("//xs:element", namespaces=ns)
    if len(root_elements) == 1:
        return StructType(parse_element(root_elements[0]))
    else:
        raise ValueError("XSD存在多个根元素，暂不支持")

测试用例（XSD示例） #

<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
    <xs:element name="Book">
        <xs:complexType>
            <xs:sequence>
                <xs:element name="Title" type="xs:string"/>
                <xs:element name="Authors">
                    <xs:complexType>
                        <xs:sequence>
                            <xs:element name="Author" type="xs:string" maxOccurs="unbounded"/>
                        </xs:sequence>
                    </xs:complexType>
                </xs:element>
                <xs:element name="PublishDate" type="xs:date"/>
            </xs:sequence>
        </xs:complexType>
    </xs:element>
</xs:schema>

问题现象 #

• 用./xs:element时：解析后的Schema仅包含Title、Authors、PublishDate，Authors内部的Author字段完全丢失
• 用.//xs:element时：Schema中会出现重复的顶层Author字段，结构混乱：

  StructType([
      StructField('Title', StringType(), True),
      StructField('Authors', StructType([StructField('Author', ArrayType(StringType()), True)]), True),
      StructField('PublishDate', StringType(), True),
      StructField('Author', ArrayType(StringType()), True)  # 不该出现的重复项
  ])
  

解决方案：按XSD结构层级递归解析 #

核心是遵循XSD的结构逻辑：元素 -> complexType -> sequence/choice/all -> 子元素，只处理当前层级的直接子元素，避免跨层级匹配。

修改后的代码 #

from lxml import etree
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, DateType

def xsd_to_spark_schema(xsd_path):
    tree = etree.parse(xsd_path)
    root = tree.getroot()
    ns = {"xs": "http://www.w3.org/2001/XMLSchema"}
    
    def map_xsd_type_to_spark(xsd_type):
        # 可扩展更多XSD类型到PySpark类型的映射
        type_map = {
            "xs:string": StringType(),
            "xs:date": DateType(),
            "xs:int": IntegerType()
        }
        return type_map.get(xsd_type, StringType())
    
    def parse_element(element):
        fields = []
        # 先获取当前元素下的complexType节点
        complex_type = element.xpath("./xs:complexType", namespaces=ns)
        if complex_type:
            ct = complex_type[0]
            # 定位complexType下的元素容器（sequence/choice/all是XSD中常见的元素容器）
            containers = ct.xpath("./xs:sequence | ./xs:choice | ./xs:all", namespaces=ns)
            for container in containers:
                # 只提取容器的直接子元素，严格控制层级
                child_elements = container.xpath("./xs:element", namespaces=ns)
                for child in child_elements:
                    name = child.get("name")
                    xsd_type = child.get("type")
                    max_occurs = child.get("maxOccurs")
                    
                    if xsd_type:
                        spark_type = map_xsd_type_to_spark(xsd_type)
                    else:
                        # 处理匿名complexType，直接递归解析当前子元素
                        spark_type = StructType(parse_element(child))
                    
                    if max_occurs == "unbounded":
                        fields.append(StructField(name, ArrayType(spark_type), nullable=True))
                    else:
                        fields.append(StructField(name, spark_type, nullable=True))
        return fields
    
    root_elements = root.xpath("//xs:element", namespaces=ns)
    if len(root_elements) == 1:
        return StructType(parse_element(root_elements[0]))
    else:
        raise ValueError("XSD存在多个根元素，暂不支持")

效果验证 #

解析测试用例XSD后，得到正确的嵌套结构：

StructType([
    StructField('Title', StringType(), True),
    StructField('Authors', StructType([
        StructField('Author', ArrayType(StringType()), True)
    ]), True),
    StructField('PublishDate', DateType(), True)
])

关键说明 #

• 避免使用.//xs:element，它会匹配所有后代元素，导致重复提取
• 不要仅用./xs:element，因为嵌套元素存在于complexType的容器节点下，不是当前元素的直接子级
• 严格按照XSD的结构层级递归，确保每个元素只在所属的父结构中被解析一次

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