如何将 JSON 对象添加到 apache spark 中的数据集/数据框答案

【问题标题】：How to add JSON Object to a Dataset/Dataframe in apache spark如何将 JSON 对象添加到 apache spark 中的数据集/数据框
【发布时间】：2018-03-31 15:30:45
【问题描述】：

我想知道是否有可能使用 Spark Dataset API 创建自定义 JSON

或 Apache Spark 提供的任何其他功能。我知道我可以使用join() 方法加入两个数据集，但我想创建自定义 JSON，其中数据集 2 即警报在我的情况下添加到数据集 1 即库存作为带有键“ALERT”的 JSON 对象。

Dataset<Row> inventory = spark.read().option("multiLine", true).option("mode", "PERMISSIVE")
                    .json("C:\\Users\\phyadavi\\LearningAndDevelopment\\\\CDXJSONMergeJob\\data1\\inventory.json");
Dataset<Row> alerts = spark.read().option("multiLine", true).option("mode", "PERMISSIVE")
                    .json("C:\\Users\\phyadavi\\LearningAndDevelopment\\\\CDXJSONMergeJob\\data1\\alert.json");

Dataset<Row> inventoryAlerts = inventory.join(alerts);
        inventoryAlerts.printSchema();

库存和警报的架构如下。

root
 |-- Equipment: struct (nullable = true)
 |    |-- items: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- collectedPid: string (nullable = true)
 |    |    |    |-- collectedSerialNum: string (nullable = true)
 |    |    |    |-- containingHwId: string (nullable = true)
 |    |    |    |-- equipmentType: string (nullable = true)
 |    |    |    |-- hwId: string (nullable = true)
 |    |    |    |-- items: array (nullable = true)
 |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |-- tagName: string (nullable = true)
 |    |    |    |    |    |-- tagValue: string (nullable = true)
 |    |    |    |-- pceMultiPid: string (nullable = true)
 |    |    |    |-- pcePhyiscalType: string (nullable = true)
 |    |    |    |-- pcePid: string (nullable = true)
 |    |    |    |-- pceProductDescription: string (nullable = true)
 |    |    |    |-- pceProductFamily: string (nullable = true)
 |    |    |    |-- pceProductType: string (nullable = true)
 |    |    |    |-- pceRuleId: string (nullable = true)
 |    |    |    |-- productDescription: string (nullable = true)
 |    |    |    |-- productFamily: string (nullable = true)
 |    |    |    |-- productId: string (nullable = true)
 |    |    |    |-- productType: string (nullable = true)
 |    |    |    |-- serialNumber: string (nullable = true)
 |    |    |    |-- snasItemType: string (nullable = true)
 |    |    |    |-- snasProductFamily: string (nullable = true)
 |    |    |    |-- snasSerialNumber: string (nullable = true)
 |    |    |    |-- snasValidationCode: string (nullable = true)
 |    |    |    |-- snasValidationSource: string (nullable = true)
 |-- LicenseActivated: struct (nullable = true)
 |    |-- items: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- count: long (nullable = true)
 |    |    |    |-- type: string (nullable = true)
 |-- NetworkElement: struct (nullable = true)
 |    |-- items: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- hostname: string (nullable = true)
 |    |    |    |-- ipAddress: string (nullable = true)
 |    |    |    |-- isManagedNe: boolean (nullable = true)
 |    |    |    |-- items: array (nullable = true)
 |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |-- tagName: string (nullable = true)
 |    |    |    |    |    |-- tagValue: string (nullable = true)
 |    |    |    |-- lastUpdateDate: long (nullable = true)
 |    |    |    |-- managedNeId: string (nullable = true)
 |    |    |    |-- managementAddress: string (nullable = true)
 |    |    |    |-- neId: string (nullable = true)
 |    |    |    |-- neName: string (nullable = true)
 |    |    |    |-- neRegistrationStatus: string (nullable = true)
 |    |    |    |-- productFamily: string (nullable = true)
 |    |    |    |-- productId: string (nullable = true)
 |    |    |    |-- productType: string (nullable = true)
 |    |    |    |-- serialNumber: string (nullable = true)
 |    |    |    |-- smartLicenseProductInstanceIdentifier: string (nullable = true)
 |    |    |    |-- smartLicenseVirtualAccountName: string (nullable = true)
 |    |    |    |-- softwareType: string (nullable = true)
 |    |    |    |-- softwareVersion: string (nullable = true)
 |    |    |    |-- systemUptime: long (nullable = true)
 |    |    |    |-- udiProductIdentifier: string (nullable = true)
 |-- Versions: struct (nullable = true)
 |    |-- items: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- lastUpdated: long (nullable = true)
 |    |    |    |-- type: string (nullable = true)
 |    |    |    |-- version: string (nullable = true)
 |-- collectorId: string (nullable = true)
 |-- generatedAt: long (nullable = true)
 |-- managedNeId: string (nullable = true)
 |-- partyId: string (nullable = true)
 |-- recordType: string (nullable = true)
 |-- sourceNeId: string (nullable = true)
 |-- sourcePartyId: string (nullable = true)
 |-- sourceSubPartyId: string (nullable = true)
 |-- wfid: string (nullable = true)

#####################################
root
 |-- collectorId: string (nullable = true)
 |-- generatedAt: long (nullable = true)
 |-- managedNeId: string (nullable = true)
 |-- neAlert: struct (nullable = true)
 |    |-- advisory: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- equipmentType: string (nullable = true)
 |    |    |    |-- headlineName: string (nullable = true)
 |    |    |    |-- hwId: string (nullable = true)
 |    |    |    |-- neId: string (nullable = true)
 |    |    |    |-- productFamily: string (nullable = true)
 |    |    |    |-- productId: string (nullable = true)
 |    |    |    |-- psirtId: long (nullable = true)
 |    |    |    |-- publicReleaseInd: string (nullable = true)
 |    |    |    |-- softwareType: string (nullable = true)
 |    |    |    |-- softwareVersion: string (nullable = true)
 |    |    |    |-- vulnerabilityReason: string (nullable = true)
 |    |    |    |-- vulnerabilityStatus: string (nullable = true)
 |    |-- fieldNotice: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- caveat: string (nullable = true)
 |    |    |    |-- distributionCode: string (nullable = true)
 |    |    |    |-- equipmentType: string (nullable = true)
 |    |    |    |-- fieldNoticeId: long (nullable = true)
 |    |    |    |-- fieldNoticeName: string (nullable = true)
 |    |    |    |-- hwId: string (nullable = true)
 |    |    |    |-- neId: string (nullable = true)
 |    |    |    |-- productFamily: string (nullable = true)
 |    |    |    |-- productId: string (nullable = true)
 |    |    |    |-- serialNumber: string (nullable = true)
 |    |    |    |-- softwareType: string (nullable = true)
 |    |    |    |-- vulnerabilityReason: string (nullable = true)
 |    |    |    |-- vulnerabilityStatus: string (nullable = true)
 |    |-- hwEoX: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- bulletinName: string (nullable = true)
 |    |    |    |-- equipmentType: string (nullable = true)
 |    |    |    |-- hardwareEoXId: long (nullable = true)
 |    |    |    |-- hwId: string (nullable = true)
 |    |    |    |-- neId: string (nullable = true)
 |    |    |    |-- productId: string (nullable = true)
 |    |-- swEoX: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- bulletinHeadline: string (nullable = true)
 |    |    |    |-- equipmentType: string (nullable = true)
 |    |    |    |-- neId: string (nullable = true)
 |    |    |    |-- productId: string (nullable = true)
 |    |    |    |-- softwareEoXId: long (nullable = true)
 |    |    |    |-- softwareType: string (nullable = true)
 |    |    |    |-- softwareVersion: string (nullable = true)
 |-- partyId: string (nullable = true)
 |-- recordType: string (nullable = true)
 |-- sourceNeId: string (nullable = true)
 |-- sourcePartyId: string (nullable = true)
 |-- sourceSubPartyId: string (nullable = true)
 |-- wfid: string (nullable = true)

【问题讨论】：

您是否尝试过select 或withColumn 在数据集中添加键“ALERT”和union？你能分别使用inventory.printSchema 和alerts.printSchema 显示inventory 和alerts 的架构吗？将它们添加到您的问题中。谢谢。
@JacekLaskowski 我尝试使用 WithColumn() ，为此我需要将警报 json 添加为对象。但是 WithColumn() 只取 lit() 作为值。
我也尝试过使用 select 但我不知道如何使用 select 将 JSON 合并为内部 JSON。

标签： java apache-spark dataset

【解决方案1】：

如果你想加入并保持一个数据集中的字段作为嵌套，你可以使用struct创建一个StructType列并加入如下

import org.apache.spark.sql.functions.udf

Dataset<Row> inventory = spark.read().option("multiLine", true).option("mode", "PERMISSIVE")
                    .json("path to json inventory");
Dataset<Row> alerts = spark.read().option("multiLine", true).option("mode", "PERMISSIVE")
                           .json("path to alerts json")
                           .select($"partyId", struct("columns").as("ALERTS"));
//column names are all the columns that you want in nested fiels with comma separated

Dataset<Row> inventoryAlerts = inventory.join(alerts);
        inventoryAlerts.printSchema();

这应该会在join 之后为您提供所需的schema。

【讨论】：