从 kinesis 到 pyspark 读取 json 时出现问题

Question

我正在尝试从 Kinesis 读取流式 JSON 数据到 PySpark.My JSON 看起来像：

{'installmentNo': '10', 'loanId': '1'}

我已经指定了架构，但是当 spark 读取数据时我得到 'null'。下面是代码片段。

from pyspark.sql.types import *
from pyspark.sql.functions import from_json

fields = [

  StructField("installmentNo", IntegerType(), True),
  StructField("loanId", IntegerType(), True)

]
pythonSchema = StructType(fields)

kinesisDf = spark.readStream \
.format("kinesis")\
.option("streamName", kinesisStreamName)\
.option("region", kinesisRegion)\
.option("initialPosition", "latest")\
.option("awsAccessKey", awsAccessKeyId)\
.option("awsSecretKey", awsSecretKey).load()

dataDevicesDF = kinesisDf.selectExpr("cast (data as STRING) my_json_data").select(from_json("my_json_data", pythonSchema).alias("yp_inst")).select("yp_inst.*")
display(dataDevicesDF)

输出：

但是，当我删除 'from_json' 部分时，我得到一个包含 JSON 字符串的列。但我想将 json 分解为特定的列并将数据作为 df。有人可以建议我更改吗？

Answer 1

架构不正确 - 您的数据是字符串，而您声明的是整数。

请将定义更改为

pythonSchema = StructType([
    StructField("installmentNo", StringType(), True),
    StructField("loanId", StringType(), True)
])

并转换输出：

from_json(
    "my_json_data", pythonSchema
).cast("struct<installmentNo: integer, loanId: integer>"))

其余代码应保持原样，但为了清晰起见，您可以明确设置选项（因为输入不是标准 JSON）：

from_json(
    "my_json_data", pythonSchema, {"allowSingleQuotes": "true"}
).cast("struct<installmentNo: integer, loanId: integer>"))

从 kinesis 到 pyspark 读取 json 时出现问题

Issue while reading json from kinesis to pyspark

apache-spark

amazon-kinesis

pyspark