使用 Spark Structured Streaming (pyspark) 从带有（模式和有效负载）的 Kafka Connect JSONConverter 消息中提取“有效负载”

Question

然而，我想要完成的正是这个问题 (Here) 的内容；就我而言，我使用的是 Python/Pyspark 而不是 Scala。

我正在尝试提取包含架构的 Kafka 连接消息的“有效负载”部分。

示例消息：

{"schema":{"type":"struct","name":"emp_table","fields":[{"field":"emp_id","type":"string"},{"field":"emp_name","type":"String"},{"field":"city","type":"string"},{"field":"emp_sal","type":"string"},{"field":"manager_name","type":"string"}]},"payload":{"emp_id":"1","emp_name":"abc","city":"NYK","emp_sal":"100000","manager_name":"xyz"}}

第 1 步 - 为“有效负载”部分定义架构：

payload_schema = StructType([
StructField("emp_id", StringType(), False),
StructField("emp_name", StringType(), True),
StructField("city", StringType(), True),
StructField("emp_sal", StringType(), True),
StructField("manager_name", StringType(), True)])

第 2 步 - 从 Kafka 读取：

df =spark.readStream.format("kafka")

第 3 步 - 从 Kafka 消息中获取消息值：

kafka_df = df.selectExpr("CAST(value AS STRING)")

第 4 步 - 仅提取“有效负载”（我卡在这里）：

    import pyspark.sql.functions as psf

    emp_df = kafka_df\
    .select(psf.from_json(psf.col('value'), payload_schema).alias("DF"))\
    .select("DF.*")

我被困在这部分，因为在将其传递给 from_json() 函数之前，我无法弄清楚如何从 JSON 字符串中提取有效负载。

注意：我知道我需要为整个消息定义完整的模式，然后才能在 from_json() 中使用它；但是；我正在尝试仅获取“有效负载” json 字符串部分。

Answer 1

你可以使用SQL函数get_json_object：

import pyspark.sql.functions as psf

kafka_df
  .select(psf.get_json_object(kafka_df['value'],"$.payload").alias('payload'))
  .select(psf.from_json(psf.col('payload'), payload_schema).alias("DF"))
  .select("DF.*")

或者，您需要为整个消息定义完整架构，然后才能在from_json 中使用它。

这意味着您的架构应该如下所示：

full_schema = StructType([
  StructField("schema", StructType([
    StructField("type", StringType(), False),
    StructField("name", StringType(), False),
    StructField("fields", StructType([
      StructField("field", StringType(), False),
      StructField("type", StringType(), False)
    ]),
  StructField("payload", StructType([
    StructField("emp_id", StringType(), False),
    StructField("emp_name", StringType(), True),
    StructField("city", StringType(), True),
    StructField("emp_sal", StringType(), True),
    StructField("manager_name", StringType(), True)
  ])
])

请仔细检查此架构定义，因为我不完全确定如何在 Python 中的架构内定义数组，但我希望这个想法很清楚。

完成后，您可以通过

选择有效负载字段

import pyspark.sql.functions as psf

    emp_df = kafka_df\
    .select(psf.from_json(psf.col('value'), full_schema).alias("DF"))\
    .select("DF.payload.*")

Answer 2

由于某种原因，我错过了 pyspark 具有 get_json_object() 函数。 在 Mike 发表评论后，我回到了 documentation，找到了我想要的东西。

这就是答案：

    kafka_df = df.selectExpr("CAST(value AS STRING)")
    payload_df = kafka_df.select(psf.get_json_object(kafka_df.value, "$.payload").alias("payload"))
    emp_df = payload_df.select(psf.from_json(psf.col('payload'), schema).alias("DF")).select("DF.*")