使用 pyspark 来自 JSON 数据的 Hive 表

Posted 2023-03-23

【中文标题】使用 pyspark 来自 JSON 数据的 Hive 表

【英文标题】：Hive table from JSON data using pyspark

【发布时间】：2020-05-13 09:30:25

【问题描述】：

我想从另一个配置单元表（放置在一个列事件数据中）的 json 数据创建一个平面配置单元表。下面是json数据结构。我已经使用稍后的视图创建了配置单元表，但现在我想使用带有一些 UDF 的 pyspark 来创建配置单元表。

'"callId":"0000000","journey":"channel":"out":"[e-mail]","outbound":"EMAIL","application":"componentId":"23456","name":"dfgt-ghy-svc","applicationReferenceId":"SRFC98756RD","servicingDetail":"offerAttributes":"id":"ADLC0110000","offerCommunicationAttributes":"id":"CFRGTV10098","status":"SUCCESS","customerInfo":"calledInAccount":"ERFCVDG9801","correlationId":"9845-sd76-sdfr87","fulfiller":"id":"DEFC1009","category":"TST","entity":"colleague","platform":"name":"v-generation","id":"37664859"'

我只需要提取 4 列，它们是 callId、correlationId、servicingDetail -offerAttributes-id、fulfiller-id

请帮我做同样的事情。

【参考方案1】：

首先，在 hive 中创建 OutputTable。

然后，使用 get_json_object() 函数 从单列表中选择数据并将该数据插入到最终表中。参考get_json_object()。

INSERT INTO table OutputTable 
SELECT 
    get_json_object(event_data,'$.callId') as callId, 
    get_json_object(event_data,'$.correlationId') as correlationId,
    get_json_object(event_data,'$.servicingDetail.offerAttributes.id') as servicingDetail_offerAttributes_id,
    get_json_object(event_data,'$.fulfiller.id') as fulfiller_id,
FROM SingleColumnTable;

第一次编辑 - pyspark 的解决方案

将单列表读入数据框（假设 df 名称为 dfSingleColumnedData），然后应用以下逻辑来获取每列数据。最后选择您需要的列。

from pyspark.sql import Row
from collections import OrderedDict

def convert_to_row(d: dict) -> Row:
    return Row(**OrderedDict(sorted(d.items())))

dfSingleColumnedData.rdd.map(convert_to_row).toDF()

【讨论】：

我想从 pyspark 代码中获取 o/p。请在 pyspark 中转换。