PySpark中使用XPath从XML字符串提取数据的正确指南-Python教程-PHP中文网

PySpark中使用XPath从XML字符串提取数据的正确指南

在使用PySpark的xpath函数从XML字符串中提取数据时，开发者常遇到提取节点文本内容时返回空值数组的问题。本文将深入解析这一常见误区，指出获取节点文本内容需明确使用text()函数，而提取属性值则直接使用@attributeName。通过详细的代码示例，本文将指导您正确地从复杂的XML结构中提取所需信息，避免常见错误，提升数据处理效率。

PySpark中XPath函数概述

pyspark提供了强大的xpath函数，允许用户利用xpath表达式从结构化的xml字符串中解析和提取数据。这对于处理包含xml格式数据的半结构化数据集至关重要。xpath函数通常以xpath(xml_string_column, xpath_expression)的形式使用，它返回一个包含匹配结果的数组。

常见问题：提取节点文本内容时出现空值数组

许多用户在使用xpath函数尝试提取XML节点的文本内容时，可能会发现结果是一个包含null值的数组，而不是期望的文本数据。例如，当尝试提取<Name>John Doe</Name>中的"John Doe"时，如果XPath表达式写为/Root/Customers/Customer/Name，PySpark的xpath函数会返回[null, null, null]等结果。

示例XML数据：

假设我们有一个包含以下XML字符串的DataFrame列：

<?xml version="1.0" encoding="utf-8"?>
<Root>
    <Customers>
        <Customer CustomerID="1">
            <Name>John Doe</Name>
            <Address>
                <Street>123 Main St</Street>
                <City>Anytown</City>
                <State>CA</State>
                <Zip>12345</Zip>
            </Address>
            <PhoneNo>123-456-7890</PhoneNo>
        </Customer>
        <Customer CustomerID="2">
            <Name>Jane Smith</Name>
            <Address>
                <Street>456 Oak St</Street>
                <City>Somecity</City>
                <State>NY</State>
                <Zip>67890</Zip>
            </Address>
            <PhoneNo>987-654-3210</PhoneNo>
        </Customer>
    </Customers>
    <Orders>
        <!-- ...其他订单数据... -->
    </Orders>
</Root>

登录后复制

错误的代码示例：

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder.appName("XML_Extraction").getOrCreate()

# 模拟从CSV读取数据，并进行初步清理
# 假设df_Customers_Orders包含一列名为"Data"的字符串，其中是上述XML
data_row = [("""<?xml version="1.0" encoding="utf-8"?>
<Root>
    <Customers>
        <Customer CustomerID="1">
            <Name>John Doe</Name>
            <Address>
                <Street>123 Main St</Street>
                <City>Anytown</City>
                <State>CA</State>
                <Zip>12345</Zip>
            </Address>
            <PhoneNo>123-456-7890</PhoneNo>
        </Customer>
        <Customer CustomerID="2">
            <Name>Jane Smith</Name>
            <Address>
                <Street>456 Oak St</Street>
                <City>Somecity</City>
                <State>NY</State>
                <Zip>67890</Zip>
            </Address>
            <PhoneNo>987-654-3210</PhoneNo>
        </Customer>
        <Customer CustomerID="3">
            <Name>Bob Johnson</Name>
            <Address>
                <Street>789 Pine St</Street>
                <City>Othercity</City>
                <State>TX</State>
                <Zip>11223</Zip>
            </Address>
            <PhoneNo>456-789-0123</PhoneNo>
        </Customer>
    </Customers>
    <Orders>
        <Order>
            <CustomerID>1</CustomerID>
            <EmpID>100</empID>
            <OrderDate>2022-01-01</OrderDate>
            <Cost>100.50</cost>
        </Order>
        <Order>
            <CustomerID>2</CustomerID>
            <EmpID>101</empID>
            <OrderDate>2022-01-02</OrderDate>
            <Cost>200.75</cost>
        </Order>
    </Orders>
</Root>"""),]
df_Customers_Orders = spark.createDataFrame(data_row, ["Data"])

# 如果XML字符串被双引号包裹，需要进行清理
# df_Customers_Orders = df_Customers_Orders.withColumn("Data", expr("substring(Data, 2, length(Data)-2)"))
# df_Customers_Orders = df_Customers_Orders.withColumn("Data", regexp_replace("Data", '""', '"'))

df_sample_CustomersOrders_incorrect = df_Customers_Orders.selectExpr(
    "xpath(Data,'/Root/Customers/Customer/@CustomerID') as CustomerID",
    "xpath(Data,'/Root/Customers/Customer/Name') as ContactName",
    "xpath(Data,'/Root/Customers/Customer/PhoneNo') as PhoneNo",
)

df_sample_CustomersOrders_incorrect.show(truncate=False)

登录后复制

输出结果：

+----------+------------------------+------------------------+
|CustomerID|ContactName             |PhoneNo                 |
+----------+------------------------+------------------------+
|[1, 2, 3] |[null, null, null, null]|[null, null, null, null]|
+----------+------------------------+------------------------+

登录后复制

可以看到，ContactName和PhoneNo列返回了null值的数组。

解决方案：使用text()函数提取节点文本内容

PySpark的xpath函数遵循标准的XPath规范。在XPath中，直接指定节点路径（如/Root/Customers/Customer/Name）通常是选择节点本身，而不是其内部的文本内容。要明确提取节点的文本内容，需要追加text()函数。

提取节点文本内容： 使用xpath_expression/text()
提取节点属性值： 使用xpath_expression/@attributeName

这解释了为什么CustomerID（通过@CustomerID提取属性）能够正确获取值，而ContactName和PhoneNo（直接指向节点）却返回空值。

怪兽AI数字人

数字人短视频创作，数字人直播，实时驱动数字人

查看详情

正确的代码示例：

修改上述代码，为需要提取文本内容的XPath表达式添加text()。

df_sample_CustomersOrders_correct = df_Customers_Orders.selectExpr(
    "xpath(Data,'/Root/Customers/Customer/@CustomerID') as CustomerID",
    "xpath(Data,'/Root/Customers/Customer/Name/text()') as ContactName",
    "xpath(Data,'/Root/Customers/Customer/PhoneNo/text()') as PhoneNo",
)

df_sample_CustomersOrders_correct.show(truncate=False)

登录后复制

输出结果：

+----------+----------------------------+----------------------------+
|CustomerID|ContactName                 |PhoneNo                     |
+----------+----------------------------+----------------------------+
|[1, 2, 3] |[John Doe, Jane Smith, Bob J.]|[123-456-7890, 987-654-3210, 456-789-0123]|
+----------+----------------------------+----------------------------+

登录后复制

现在，ContactName和PhoneNo列都正确地提取出了相应的文本内容。

完整示例与注意事项

在实际应用中，您可能需要将提取出的数组展开成多行，或者进一步处理这些数据。

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import StringType

spark = SparkSession.builder.appName("XML_Extraction_Tutorial").getOrCreate()

# 模拟包含XML字符串的CSV文件
# 通常，如果CSV文件中XML字符串被双引号包裹，或有转义字符，需要预处理
# 这里直接创建DataFrame以简化示例，但在实际中，read.csv后可能需要以下清理步骤：
# df_Customers_Orders = spark.read.option("header", "true").csv("source.csv")
# df_Customers_Orders = df_Customers_Orders.withColumn("Data", expr("substring(Data, 2, length(Data)-2)"))
# df_Customers_Orders = df_Customers_Orders.withColumn("Data", regexp_replace("Data", '""', '"'))

xml_string = """<?xml version="1.0" encoding="utf-8"?>
<Root>
    <Customers>
        <Customer CustomerID="1">
            <Name>John Doe</Name>
            <Address>
                <Street>123 Main St</Street>
                <City>Anytown</City>
                <State>CA</State>
                <Zip>12345</Zip>
            </Address>
            <PhoneNo>123-456-7890</PhoneNo>
        </Customer>
        <Customer CustomerID="2">
            <Name>Jane Smith</Name>
            <Address>
                <Street>456 Oak St</Street>
                <City>Somecity</City>
                <State>NY</State>
                <Zip>67890</Zip>
            </Address>
            <PhoneNo>987-654-3210</PhoneNo>
        </Customer>
        <Customer CustomerID="3">
            <Name>Bob Johnson</Name>
            <Address>
                <Street>789 Pine St</Street>
                <City>Othercity</City>
                <State>TX</State>
                <Zip>11223</Zip>
            </Address>
            <PhoneNo>456-789-0123</PhoneNo>
        </Customer>
    </Customers>
    <Orders>
        <Order>
            <CustomerID>1</CustomerID>
            <EmpID>100</empID>
            <OrderDate>2022-01-01</OrderDate>
            <Cost>100.50</cost>
        </Order>
        <Order>
            <CustomerID>2</CustomerID>
            <EmpID>101</empID>
            <OrderDate>2022-01-02</OrderDate>
            <Cost>200.75</cost>
        </Order>
    </Orders>
</Root>"""

df_xml_data = spark.createDataFrame([(xml_string,)], ["Data"])
df_xml_data.show(truncate=False)

# 使用xpath函数提取数据
df_extracted_customers = df_xml_data.selectExpr(
    "xpath(Data,'/Root/Customers/Customer/@CustomerID') as CustomerID_Array",
    "xpath(Data,'/Root/Customers/Customer/Name/text()') as ContactName_Array",
    "xpath(Data,'/Root/Customers/Customer/PhoneNo/text()') as PhoneNo_Array",
)

df_extracted_customers.show(truncate=False)

# 将数组列展开成多行，以便于后续处理
# 这里假设所有数组的长度相同，或者您只关心匹配到的第一个元素
df_flattened_customers = df_extracted_customers.select(
    explode("CustomerID_Array").alias("CustomerID"),
    explode("ContactName_Array").alias("ContactName"),
    explode("PhoneNo_Array").alias("PhoneNo")
)

df_flattened_customers.show(truncate=False)

# 写入CSV文件
# df_flattened_customers.write.format("csv").option("header", "true").mode("overwrite").save("path_to_output.csv")

spark.stop()

登录后复制

注意事项：

XPath表达式的精确性： 确保您的XPath表达式准确无误地指向目标节点或属性。错误的路径会导致空数组或不正确的结果。
text()的重要性： 牢记提取节点文本内容时必须使用text()，而提取属性值时使用@attributeName。这是最常见的错误源。
返回类型： xpath函数总是返回一个ArrayType(StringType)。如果您的XML中有多个匹配项，它们将全部作为字符串存储在这个数组中。如果只有一个匹配项，数组中将只有一个元素。
explode函数： 当xpath返回一个数组，且您希望将数组中的每个元素作为单独的行进行处理时，可以使用explode函数将数组列展平。
XML预处理： 如果XML字符串是从外部源（如CSV文件）读取的，它可能被双引号包裹或包含转义字符。您可能需要使用substring、regexp_replace等函数进行清理，确保XML字符串是有效的。
性能考虑： 对于非常大的XML字符串或包含大量XML数据的DataFrame，频繁使用xpath函数可能会有性能开销。考虑是否可以在数据摄取阶段就进行XML解析，或者评估其他更专业的XML解析库（如spark-xml，如果整个列都是XML）。

总结

在PySpark中使用xpath函数从XML字符串中提取数据是一个常见的操作。理解XPath表达式中节点文本内容（text()）与属性值（@attributeName）的提取差异是避免空值数组的关键。通过本文提供的指南和代码示例，您可以更高效、准确地处理XML数据，从而避免常见的陷阱，确保数据提取的正确性。

以上就是PySpark中使用XPath从XML字符串提取数据的正确指南的详细内容，更多请关注php中文网其它相关文章！