在Pyspark数据框中解析XML列

如何解决在Pyspark数据框中解析XML列

我是PySpark的新手，正在尝试解决数据问题。我有一个pyspark DF，它是使用从MS SQL Server中提取的数据创建的，具有2列：ID（整数）和XMLMsg（字符串）。第二列XMLMsg包含XML格式的数据。 目标是解析XMLMsg列，并使用从XML中提取的列在同一DF中创建其他列。

以下是pyspark DF的示例结构：

.rtf

预期输出为：

ID  XMLMsg
101 ...<a><b>name1</b><c>loc1</c></a>...<d>dept1</d>...
102 ...<a><b>name2</b><c>loc2</c></a>...<d>dept2</d>...
103 ...<a><b>name3</b><c>loc3</c></a>...<d>dept3</d>...

根据对SO的搜索，我尝试了一些建议；但是，无法达到预期的结果。因此，请寻求帮助和指导。谢谢您的时间。

解决方法

我最终使用Lambda和UDF解决了这个问题，因为我不得不从一个巨大的XML文件中从4个节点中获取文本。由于XML文件已经存在于pyspark Dataframe的一列和一部分中，所以我不想编写为文件并再次解析整个XML。我也想避免使用XSD模式。 实际的xml有多个命名空间，还有一些具有特定条件的节点。 示例：

<ap:applicationproduct xmlns:xsi="http://www.example.com/2005/XMLSchema-instance" xmlns:ap="http://example.com/productinfo/1_6" xmlns:ct="http://example.com/commontypes/1_0" xmlns:dc="http://example.com/datacontent/1_0" xmlns:tp="http://aexample.com/prmvalue/1_0" ....." schemaVersion="..">
  <ap:ParameterInfo>
    <ap:Header>
      <ct:Version>1.0</ct:Version>
      <ct:Sender>ABC</ct:Sender>
      <ct:SenderVersion />
      <ct:SendTime>...</ct:SendTime>
    </ap:Header>
    <ap:ProductID>
      <ct:Model>
        <ct:Series>34AP</ct:Series>
        <ct:ModelNo>013780</ct:ModelNo>
     ..............
      ..............
   <ap:Object>
        <ap:Parameter schemaVersion="2.5" Code="DDA">
          <dc:Value>
            <tp:Blob>mbQAEAgBTgKQEBAX4KJJYABAIASL0AA==</tp:Blob>
          </dc:Value>
        </ap:Parameter>
.........
........

在这里，我需要从ct：ModelNo和tp：Blob中提取值

from pyspark.sql.types import *
from pyspark.sql.functions import udf
import xml.etree.ElementTree as ET
# List of namespaces to be used:
ns = {'ap' : 'http://example.com/productinfo/1_6','ct':'http://example.com/commontypes/1_0','dc':'http://example.com/datacontent/1_0','tp':'http://aexample.com/prmvalue/1_0' 
    }

parsed_model = (lambda x:    ET.fromstring(x).find('ap:ParameterInfo/ap:ProductID/ct:Model/ct:ModelNo',ns).text)
udf_model = udf(parsed_model)
parsed_model_df = df.withColumn('ModelNo',udf_Model('XMLMsg'))

对于具有blob值的节点，也可以编写类似的函数，但该节点的路径为： 'ap：ParameterInfo / ap：Object / ap：Parameter [@ Code =“ DDA”] / dc：Value / tp：Blob'

这对我有用，我能够在pyspark DF中添加所需的值。欢迎任何建议，以使它更好。谢谢！