Pyspark如何解析单行多JSON文件并关联匹配的事件数据

现有单行JSON数据，单条行内拼接了多个独立JSON对象，示例数据如下：

{"test":"valid2","workflowId":79370,"email":"d1@example.com"}{"email":"d1@example.com","eventName":"emailOpen","dataFields":{"campaignId":1125010,"ip":"100.100.200.243","userAgentDevice":"Gmail","messageId":"be4e071c11594bb0b4ee3c444fd08b99","emailId":"d1@example.com","userAgent":"Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0 (via ggpht.com GoogleImageProxy)","workflowName":"DWH TEST 06042020 WITH CALL","locale":null,"templateId":1576122,"emailSubject":"DWH TEST","labels":[],"createdAt":"2020-04-06 15:06:16 +00:00","templateName":"DWH TEST","messageTypeId":27043,"experimentId":79413,"campaignName":"DWH Test Automation","workflowId":79370,"email":"d1@example.com","channelId":24365}}{"email":"dd1@example.com","eventName":"emailOpen","dataFields":{"campaignId":1100,"ip":"50.100.200.243","userAgentDevice":"Gmail","messageId":"zz4e071c11594bb0b4ee3c444fd08b99","emailId":"dd1@example.com","userAgent":"Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0 (via ggpht.com GoogleImageProxy)","workflowName":"TEST","locale":null,"templateId":1576122,"emailSubject":"TEST","labels":"Cambbridge test","createdAt":"2020-04-10 15:06:16 +00:00","templateName":"TEST","messageTypeId":27043,"experimentId":89413,"campaignName":"Cambridge Test","workflowId":18370,"email":"dd1@example.com","channelId":1111}}{"email":"dd1@example.com","eventName":"emailClick","dataFields":{"campaignId":1100,"ip":"50.100.200.243","userAgentDevice":"Gmail","messageId":"zzee071c11594bb0b4ee3c444fd08b99","emailId":"dd1@example.com","userAgent":"Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0 (via ggpht.com GoogleImageProxy)","workflowName":"TEST","locale":null,"templateId":1576122,"emailSubject":"TEST","labels":"Cambbridge test","createdAt":"2020-04-10 15:08:16 +00:00","templateName":"TEST","messageTypeId":27043,"experimentId":89413,"campaignName":"Cambridge Test","workflowId":18370,"email":"dd1@example.com","channelId":1111}}{"test":"valid2","workflowId":79370,"email":"d1@example.com"}{"email":"d1@example.com","eventName":"emailOpen","dataFields":{"campaignId":1125010,"ip":"100.100.200.243","userAgentDevice":"Gmail","messageId":"be4e071c11594bb0b4ee3c444fd08b99","emailId":"d1@example.com","userAgent":"Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0 (via ggpht.com GoogleImageProxy)","workflowName":"DWH TEST 06042020 WITH CALL","locale":null,"templateId":1576122,"emailSubject":"DWH TEST","labels":[],"createdAt":"2020-04-06 15:06:16 +00:00","templateName":"DWH TEST","messageTypeId":27043,"experimentId":79413,"campaignName":"DWH Test Automation","workflowId":79370,"email":"d1@example.com","channelId":24365}}

数据规则：

• 包含test、workflowId、email三个字段的为附加JSON对象
• 后续跟的所有workflowId、email与附加JSON匹配的为事件JSON对象，需要和对应的附加JSON关联
• 单行内可存在多组「附加JSON+对应事件JSON」的结构

直接使用Pyspark默认读JSON的代码只能读到第一个附加JSON的数据：

df = sql_context.read.json('test.json')
df.show() 

输出结果不符合要求：

+--------------+------+----------+
|         email|  test|workflowId|
+--------------+------+----------+
|c1@example.com|valid1|     79370|
+--------------+------+----------+

最终需要得到如下格式的DataFrame：

id                              email           event     workflow_id   custom  createdatdate   createdattime
0   be4e071c11594bb0b4ee3c444fd08b99    d1@example.com  emailOpen   79370       valid2  2020414         154248
1   be4e071c11594bb0b4ee3c444fd08b99    d1@example.com  emailOpen   79370       valid2  2020414         154248

Pyspark默认读JSON要求每行一个完整JSON对象，所以需要先按文本读取、拆分单行为多个JSON后再解析，完整实现如下：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, split, explode, from_json, schema_of_json, lit, when, last, to_timestamp, date_format, regexp_replace
from pyspark.sql.window import Window

spark = SparkSession.builder.appName("SplitMultiJson").getOrCreate()
# 1. 按文本读取源文件，整行内容存为单列
raw_df = spark.read.text("test.json")

# 2. 拆分单行内拼接的多个JSON对象
split_df = raw_df.withColumn("json_arr", split(regexp_replace(col("value"), "}\\{", "}###{"), "###"))
# 炸开数组，一行对应一个JSON字符串
exploded_df = split_df.select(explode(col("json_arr")).alias("json_str"))

# 3. 定义两种JSON的schema并解析
# 附加JSON schema
attach_schema = schema_of_json('{"test":"valid2","workflowId":79370,"email":"d1@example.com"}')
# 事件JSON schema
event_schema = schema_of_json('{"email":"d1@example.com","eventName":"emailOpen","dataFields":{"campaignId":1125010,"ip":"100.100.200.243","userAgentDevice":"Gmail","messageId":"be4e071c11594bb0b4ee3c444fd08b99","emailId":"d1@example.com","userAgent":"Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0 (via ggpht.com GoogleImageProxy)","workflowName":"DWH TEST 06042020 WITH CALL","locale":null,"templateId":1576122,"emailSubject":"DWH TEST","labels":[],"createdAt":"2020-04-06 15:06:16 +00:00","templateName":"DWH TEST","messageTypeId":27043,"experimentId":79413,"campaignName":"DWH Test Automation","workflowId":79370,"email":"d1@example.com","channelId":24365}}')

parsed_df = exploded_df.withColumn("attach_data", from_json(col("json_str"), attach_schema)) \
    .withColumn("event_data", from_json(col("json_str"), event_schema))

# 4. 关联附加JSON和对应事件JSON
# 新增行号保证JSON顺序不变
ordered_df = parsed_df.withColumn("row_num", monotonically_increasing_id())

# 用窗口函数向前填充附加JSON的字段值
w_fill = Window.orderBy("row_num").rowsBetween(Window.unboundedPreceding, 0)
filled_df = ordered_df.withColumn("custom", when(col("attach_data.test").isNotNull(), col("attach_data.test"))) \
    .withColumn("custom", last("custom", ignorenulls=True).over(w_fill)) \
    .withColumn("attach_workflow_id", when(col("attach_data.workflowId").isNotNull(), col("attach_data.workflowId"))) \
    .withColumn("attach_workflow_id", last("attach_workflow_id", ignorenulls=True).over(w_fill)) \
    .withColumn("attach_email", when(col("attach_data.email").isNotNull(), col("attach_data.email"))) \
    .withColumn("attach_email", last("attach_email", ignorenulls=True).over(w_fill))

# 5. 过滤事件JSON并提取目标字段
result_df = filled_df.filter(col("event_data.eventName").isNotNull()) \
    .filter((col("event_data.workflowId") == col("attach_workflow_id")) & (col("event_data.email") == col("attach_email"))) \
    .select(
        col("event_data.dataFields.messageId").alias("id"),
        col("event_data.email").alias("email"),
        col("event_data.eventName").alias("event"),
        col("event_data.workflowId").alias("workflow_id"),
        col("custom"),
        to_timestamp(col("event_data.dataFields.createdAt"), "yyyy-MM-dd HH:mm:ss Z").alias("created_at"),
    ) \
    .withColumn("createdatdate", date_format(col("created_at"), "yyyyMMdd").cast("int")) \
    .withColumn("createdattime", date_format(col("created_at"), "HHmmss").cast("int")) \
    .drop("created_at")

# 输出结果
result_df.show(truncate=False)

内容的提问来源于stack exchange，提问作者Rasika