解析嵌套 XML 并将数据展平为行 - PySpark

问：

我有一个复杂的 xml 文件，需要使用 PySpark 解析和展平。我将利用 AWS Glue 和 Spark 框架来完成此任务。我可以将我的 xml 文件转换为 spark 数据帧，但我需要展平数据。是否可以将PRVDR_INFO和注册分成 2 个不同的数据帧，以便我可以使用代理键将它们链接在一起？检查下面的输出。

您将在下面找到我的嵌套 XML 文件。此 XML 文件有 2 个顶级元素，分别是PRVDR_INFO和注册。

<PROVIDER>
    <PRVDR_INFO>
        <INDVDL_INFO>
            <BIRTH_DT>19831222</BIRTH_DT>
            <BIRTH_STATE_CD>VA</BIRTH_STATE_CD>
            <BIRTH_STATE_NAME>VIRGINIA</BIRTH_STATE_NAME>
            <BIRTH_CNTRY_CD>US</BIRTH_CNTRY_CD>
            <BIRTH_CNTRY_NAME>UNITED STATES</BIRTH_CNTRY_NAME>
            <BIRTH_FRGN>X</BIRTH_FRGN>
            <NAME_LIST>
                <INDVDL_NAME>
                    <NAME_CD>I</NAME_CD>
                    <NAME_DESC>INDIVIDUAL NAME</NAME_DESC>
                    <FIRST_NAME>LEO</FIRST_NAME>
                    <LAST_NAME>MESSI</LAST_NAME>
                    <TRMNTN_DT>2010-12-27T09:43:18.000000000</TRMNTN_DT>
                    <DATA_STUS_CD>HISTORY</DATA_STUS_CD>
                </INDVDL_NAME>
                <INDVDL_NAME>
                    <NAME_CD>I</NAME_CD>
                    <NAME_DESC>INDIVIDUAL NAME</NAME_DESC>
                    <FIRST_NAME>LEO</FIRST_NAME>
                    <MDL_NAME>A</MDL_NAME>
                    <LAST_NAME>WHITE</LAST_NAME>
                    <DATA_STUS_CD>CURRENT</DATA_STUS_CD>
                </INDVDL_NAME>
            </NAME_LIST>
            <XX_DEA>
                <DEA_NUM>XX0919969</DEA_NUM>
                <EFCTV_DT>20030103</EFCTV_DT>
                <DATA_STUS_CD>CURRENT</DATA_STUS_CD>
            </XX_DEA>
        </INDVDL_INFO>
    </PRVDR_INFO>
    <ENROLMENT>
        <ABC_999>
            <ENRLMT_INFO>
                <ENRLMT_DTLS>
                    <FORM_TYPE_CD>1111</FORM_TYPE_CD>
                    <ENRLMT_ID>I3994444141</ENRLMT_ID>
                    <ENRLMT_STUS_DLTS>
                        <STUS_CD>06</STUS_CD>
                        <STUS_DESC>APPROVED</STUS_DESC>
                        <STUS_DT>2019-09-25T14:11:08.0000000</STUS_DT>
                        <DATA_STUS_CD>CURRENT</DATA_STUS_CD>
                        <ENRLMT_STUS_RSN>
                            <STUS_RSN_CD>048</STUS_RSN_CD>
                            <STUS_XXX_DESC>APPROVED AFTER 2nd CONTACT</STUS_XXX_DESC>
                            <DATA_STUS_CD>CURRENT</DATA_STUS_CD>
                        </ENRLMT_STUS_RSN>
                    </ENRLMT_STUS_DLTS>
                </ENRLMT_DTLS>
            </ENRLMT_INFO>
            <PEC_SGNTR>
                <CRTFCTN_SGNTR_DT>20101109</CRTFCTN_SGNTR_DT>
                <FIRST_NAME>MIKE</FIRST_NAME>
                <LAST_NAME>BLACK</LAST_NAME>
                <TIN>555669999</TIN>
                <TAX_IDENT_TYPE_CD>S</TAX_IDENT_TYPE_CD>
                <SGNTR_EFCTV_DT>20101109</SGNTR_EFCTV_DT>
                <SGNTR_STUS_CD>9</SGNTR_STUS_CD>
                <DATA_STUS_CD>CURRENT</DATA_STUS_CD>
            </PEC_SGNTR>
        </ABC_999>
    </ENROLMENT>
</PROVIDER>

下面的代码给了我 2 列嵌套数据，但我需要将这些数据展平为多行。但是，由于输出注册和PRVDR_INFO是最重要的元素，因此我想将它们分成 2 个不同的框架，并使用 FK 将它们链接起来。

电流输出：

+--------------------+--------------------+
|           ENROLMENT|          PRVDR_INFO|
+--------------------+--------------------+
|{{{{I3994444141, ...|{{US, UNITED STAT...|
+--------------------+--------------------+

代码如下：

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args["JOB_NAME"], args)

try:
    rowTag = "PROVIDER"

    # Script generated for node S3 bucket
    df = glueContext.create_dynamic_frame.from_options(
    format_options={"rowTag": "PROVIDER"},
    connection_type="s3",
    format="xml",
    connection_options={
        "paths": ["s3://aws-glue-bucket/xml_files/small_xml.xml"]
    })
    
    print(" *********** Printing schema *************")
    df.printSchema()
    
    print(" *********** Printing Data *************")
    df.show(5)
    
    df1 = df.toDF();
    df1.show(8);
    
except Exception as glue_exception_error:
    print("##################### -- Error: "  + str(glue_exception_error) + " -- ##########################")
    raise

提前致谢。