Добавление части родительского столбца схемы к дочернему элементу во вложенном json в кадре данных spark

У меня есть ниже XML, который я пытаюсь загрузить, чтобы зажечь фрейм данных.

   <?xml version="1.0"?>
<env:ContentEnvelope xsi:schemaLocation="http">
    <env:Header>
        <env:Info>
            <env:Id>urn:uuid:6d2af93bfbfc49da9805aebb6a38996d</env:Id>
            <env:TimeStamp>20171122T07:56:09+00:00</env:TimeStamp>
        </env:Info>
        <fun:OrgId>18227</fun:OrgId>
        <fun:DataPartitionId>1</fun:DataPartitionId>
    </env:Header>
    <env:Body minVers="0.0" majVers="1" contentSet="Fundamental">
        <env:ContentItem action="Overwrite">
            <env:Data xsi:type="sr:FinancialSourceDataItem">
                <sr:Source sourceId="344" organizationId="4295906830">
                    <sr:FilingDateTime>20171111T17:00:00+00:00</sr:FilingDateTime>
                    <sr:SourceTypeCode>10K</sr:SourceTypeCode>
                    <sr:StatementDate>20171030T00:00:00+00:00</sr:StatementDate>
                    <sr:IsFilingDateTimeEstimated>false</sr:IsFilingDateTimeEstimated>
                    <sr:ContainsPreliminaryData>false</sr:ContainsPreliminaryData>
                    <sr:CapitalChangeAdjustmentDate>20171030T00:00:00+00:00</sr:CapitalChangeAdjustmentDate>
                    <sr:CumulativeAdjustmentFactor>1.00000</sr:CumulativeAdjustmentFactor>
                    <sr:ContainsRestatement>false</sr:ContainsRestatement>
                    <sr:FilingDateTimeUTCOffset>300</sr:FilingDateTimeUTCOffset>
                    <sr:ThirdPartySourceCode>SS</sr:ThirdPartySourceCode>
                    <sr:ThirdPartySourcePriority>1</sr:ThirdPartySourcePriority>
                    <sr:Auditors>
                        <sr:Auditor auditorId="3541">
                            <sr:AuditorOpinionCode>UNQ</sr:AuditorOpinionCode>
                            <sr:IsPlayingAuditorRole>true</sr:IsPlayingAuditorRole>
                            <sr:IsPlayingTaxAdvisorRole>false</sr:IsPlayingTaxAdvisorRole>
                            <sr:AuditorEnumerationId>3024068</sr:AuditorEnumerationId>
                            <sr:AuditorOpinionId>3010546</sr:AuditorOpinionId>
                            <sr:IsPlayingCSRAuditorRole>false</sr:IsPlayingCSRAuditorRole>
                        </sr:Auditor>
                        <sr:Auditor auditorId="9574">
                            <sr:AuditorOpinionCode>UWE</sr:AuditorOpinionCode>
                            <sr:IsPlayingAuditorRole>true</sr:IsPlayingAuditorRole>
                            <sr:IsPlayingTaxAdvisorRole>false</sr:IsPlayingTaxAdvisorRole>
                            <sr:AuditorEnumerationId>3030421</sr:AuditorEnumerationId>
                            <sr:AuditorOpinionId>3010547</sr:AuditorOpinionId>
                            <sr:IsPlayingCSRAuditorRole>false</sr:IsPlayingCSRAuditorRole>
                        </sr:Auditor>
                    </sr:Auditors>
                    <sr:SourceTypeId>3011835</sr:SourceTypeId>
                    <sr:ThirdPartySourceCodeId>1000716240</sr:ThirdPartySourceCodeId>
                </sr:Source>
            </env:Data>
        </env:ContentItem>
    </env:Body>
</env:ContentEnvelope>

Основной тег <env:ContentEnvelope>Тогда есть две части один заголовок (<env:Header>) и прочее это тело (<env:Body)

Детали в теле, как <fun:OrgId> а также <fun:DataPartitionId> будет одинаковым для всех строк в <env:Body,

Из этого я хочу создать два кадра данных.

Один для <sr:Source и второй для <sr:Auditor

Для обоих кадров данных action="Overwrite" будет таким же, как общий столбец.

Также из-за <sr:Auditor это ребенок <sr:Source так мало столбцов, как sourceId="344" organizationId="4295906830" будет повторяться в <sr:Auditor фрейм данных.

Это то, что я сделал до сих пор, чтобы достичь этого

    val sqlContext = new org.apache.spark.sql.SQLContext(sc)

    val dfContentEnvelope = sqlContext.read.format("com.databricks.spark.xml").option("rowTag", "env:ContentEnvelope").load("s3://trfsmallfffile/XML")

    val dfHeader = dfContentEnvelope.withColumn("Header", (dfContentEnvelope("env:Header"))).select("Header.*")
    val dfDataPartitionId =dfHeader.select("fun:DataPartitionId")
    //dfDataPartitionId.show()

   //val dfBody = sqlContext.read.format("com.databricks.spark.xml").option("rowTag", "env:Body").load("s3://trfsmallfffile/XML")
    val dfContentItem = dfContentEnvelope.withColumn("column1", explode(dfContentEnvelope("env:Body.env:ContentItem"))).select("column1.*")
    val dfType=dfContentItem.select("env:Data.*")
    //dfType.show()

      val srSource = dfType.withColumn("srSource", (dfType("sr:Source"))).select("srSource.*").drop("sr:Auditors").filter($"srSource".isNotNull)
     val srSourceAuditor = dfType.withColumn("srSource", explode(dfType("sr:Source.sr:Auditors.sr:Auditor"))).select("srSource.*")

Итак, мой вопрос, как я могу получить родительский фрейм данных для <sr:Source и дочерний фрейм для <sr:Auditor с organizationId и sourceId от родительского к дочернему фрейму данных?

1 ответ

Решение

Если вы хотите получить два кадра данных: один для Source и один для Auditors с organizationId а также sourceId из Source Dataframe, то вы можете использовать следующую логику.

Наблюдая за данными и вашими попытками, я могу предложить explode функция на env:Body.env:ContentItem столбец даст вам родительский фрейм данных

import sqlContext.implicits._
import org.apache.spark.sql.functions._
val dfContentEnvelope = sqlContext.read.format("com.databricks.spark.xml")
  .option("rowTag", "env:ContentEnvelope")
  .load("s3://trfsmallfffile/XML")

val dfContentItem = dfContentEnvelope.withColumn("column1", explode(dfContentEnvelope("env:Body.env:ContentItem"))).select("column1.*")
val ParentDF=dfContentItem.select($"env:Data.sr:Source._organizationId".as("organizationId"), $"env:Data.sr:Source._sourceId".as("sourceId"), $"env:Data.sr:Source".as("Source"))

что бы дать вам

+--------------+--------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|organizationId|sourceId|Source                                                                                                                                                                                                                                                 |
+--------------+--------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|4295906830    |344     |[4295906830,344,[WrappedArray([3541,3024068,UNQ,3010546,true,false,false], [9574,3030421,UWE,3010547,true,false,false])],20171030T00:00:00+00:00,false,false,1.0,20171111T17:00:00+00:00,300,false,10K,3011835,20171030T00:00:00+00:00,SS,1000716240,1]|
+--------------+--------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

И для дочернего кадра данных вам нужно будет взорвать sr:Auditor из вышеуказанного родительского кадра данных как

val childDF=ParentDF.select($"organizationId", $"sourceId", explode($"Source.sr:Auditors.sr:Auditor").as("Auditors"))

который должен дать вам

+--------------+--------+-------------------------------------------+
|organizationId|sourceId|Auditors                                   |
+--------------+--------+-------------------------------------------+
|4295906830    |344     |[3541,3024068,UNQ,3010546,true,false,false]|
|4295906830    |344     |[9574,3030421,UWE,3010547,true,false,false]|
+--------------+--------+-------------------------------------------+

Я надеюсь, что ответ полезен

Другие вопросы по тегам