我對(duì) PySpark 比較陌生，正在嘗試解決數(shù)據(jù)問題。我有一個(gè) pyspark DF，它是用從 MS SQL Server 中提取的數(shù)據(jù)創(chuàng)建的，有 2 列：ID（整數(shù)）和 XMLMsg（字符串）。第 2 列 XMLMsg 包含 XML 格式的數(shù)據(jù)。目標(biāo)是解析 XMLMsg 列并在同一個(gè) DF 中使用從 XML 中提取的列創(chuàng)建其他列。以下是 pyspark DF 的示例結(jié)構(gòu)：ID XMLMsg101 ...<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>...預(yù)期輸出是：ID XMLMsg b c d101 ...<a><b>name1</b><c>loc1</c></a>...<d>dept1</d>... name1 loc1 dept1102 ...<a><b>name2</b><c>loc2</c></a>...<d>dept2</d>... name2 loc2 dept2103 ...<a><b>name3</b><c>loc3</c></a>...<d>dept3</d>... name3 loc3 dept3根據(jù)我在 SO 中的搜索，我嘗試了一些建議；然而，未能達(dá)到預(yù)期的效果。因此，尋求一些幫助和指導(dǎo)。謝謝你的時(shí)間。
1 回答
largeQ
TA貢獻(xiàn)2039條經(jīng)驗(yàn) 獲得超8個(gè)贊
考慮到我必須從一個(gè)巨大的 XML 文件中獲取來自 4 個(gè)節(jié)點(diǎn)的文本，我最終使用 Lambda 和 UDF 解決了這個(gè)問題。由于 XML 文件已經(jīng)在列中并且是 pyspark Dataframe 的一部分，我不想寫成文件并再次解析整個(gè) XML。我還想避免使用 XSD 架構(gòu)。實(shí)際的 xml 有多個(gè)命名空間和一些具有特定條件的節(jié)點(diǎn)。例子：
<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'))
同樣對(duì)于具有 blob 值的節(jié)點(diǎn)，可以編寫類似的函數(shù)，但節(jié)點(diǎn)的路徑將是：'ap:ParameterInfo/ap:Object/ap:Parameter[@Code="DDA"]/dc:Value/tp:Blob'
這對(duì)我有用，我能夠在 pyspark DF 中添加所需的值。歡迎提出任何建議以使其變得更好。謝謝你！
添加回答
舉報(bào)
0/150
提交
取消