从 kafka 代理到 python 中的 Spark 流应用程序的 json 数据的正确架构是什么

Question

执行此程序后，我在数据帧的所有字段中收到空值。如果架构是问题，请向我建议此 json 格式的正确架构，否则请为我提供从 kafka-topic 读取此 json 数据的正确答案使用 readStream api 并将其打印到控制台。 这是 kafka-topic 程序中没有错误的示例 json 数据

        {
          "invoice_no": 154132541653705,
          "country": "United Kingdom",
          "timestamp": "2020-09-18 10:55:23",
          "type": "ORDER",
          "items": [
            {
              "SKU": "21485",
              "title": "RETROSPOT HEART HOT WATER BOTTLE",
              "unit_price": 4.95,
              "quantity": 6
            },
            {
              "SKU": "23499",
              "title": "SET 12 VINTAGE DOILY CHALK",
              "unit_price": 0.42,
              "quantity": 2
            }
          ]  
        }

，我在这里发布主要部分

        spark = SparkSession.builder.appName('retaildata').getOrCreate()
        spark.sparkContext.setLogLevel("ERROR")
            jsonSchema=StructType([ \
            StructField("invoice_no",IntegerType()), \
            StructField("country",StringType()), \
            StructField("timestamp",TimestampType()), \
            StructField("type",StringType()), \
            StructField("items",StructType([StructField("SKU",IntegerType()), \
                                                    StructField("title",StringType()), \
                                                    StructField("unit_price",DoubleType()), \
                                                    StructField("quantity",IntegerType()) \
                                                     ])) \
                                  ])
        
        dfraw = spark \
          .readStream \
          .format("kafka") \
          .option("kafka.bootstrap.servers", "18.211.252.152:9092") \
          .option("subscribe", "real-time-project") \
          .option("includeHeaders","true") \
          .load() \
          .select(from_json(col("value").cast("string"),jsonSchema).alias("data")).select("data.*") \
          .writeStream \
          .format("console") \
          .start() \
          .awaitTermination() \
        
        

输出：

        =================================
        I am receiving the data as
        
        Batch: 0
        -------------------------------------------
        +----------+-------+---------+----+-----+
        |invoice_no|country|timestamp|type|items|
        +----------+-------+---------+----+-----+
        +----------+-------+---------+----+-----+
        
        -------------------------------------------
        Batch: 1
        -------------------------------------------
        +----------+-------+---------+----+-----+
        |invoice_no|country|timestamp|type|items|
        +----------+-------+---------+----+-----+
        |      null|   null|     null|null| null|
        |      null|   null|     null|null| null|
        |      null|   null|     null|null| null|
        |      null|   null|     null|null| null|
        +----------+-------+---------+----+-----+

Answer 1

您只需读取 json 字符串并将其转换为 DataFrame，然后打印模式即可。

代码：

json_data = """
{
          "invoice_no": 154132541653705,
          "country": "United Kingdom",
          "timestamp": "2020-09-18 10:55:23",
          "type": "ORDER",
          "items": [
            {
              "SKU": "21485",
              "title": "RETROSPOT HEART HOT WATER BOTTLE",
              "unit_price": 4.95,
              "quantity": 6
            },
            {
              "SKU": "23499",
              "title": "SET 12 VINTAGE DOILY CHALK",
              "unit_price": 0.42,
              "quantity": 2
            }
          ]  
        }
"""

df = spark.read.json(spark.sparkContext.parallelize([json_data]))

print(df.schema.json())
print(df.schema)
df.printSchema()
df.show(10, False)

输出：

{"fields":[{"metadata":{},"name":"country","nullable":true,"type":"string"},{"metadata":{},"name":"invoice_no","nullable":true,"type":"long"},{"metadata":{},"name":"items","nullable":true,"type":{"containsNull":true,"elementType":{"fields":[{"metadata":{},"name":"SKU","nullable":true,"type":"string"},{"metadata":{},"name":"quantity","nullable":true,"type":"long"},{"metadata":{},"name":"title","nullable":true,"type":"string"},{"metadata":{},"name":"unit_price","nullable":true,"type":"double"}],"type":"struct"},"type":"array"}},{"metadata":{},"name":"timestamp","nullable":true,"type":"string"},{"metadata":{},"name":"type","nullable":true,"type":"string"}],"type":"struct"}


StructType(List(StructField(country,StringType,true),StructField(invoice_no,LongType,true),StructField(items,ArrayType(StructType(List(StructField(SKU,StringType,true),StructField(quantity,LongType,true),StructField(title,StringType,true),StructField(unit_price,DoubleType,true))),true),true),StructField(timestamp,StringType,true),StructField(type,StringType,true)))


root
 |-- country: string (nullable = true)
 |-- invoice_no: long (nullable = true)
 |-- items: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- SKU: string (nullable = true)
 |    |    |-- quantity: long (nullable = true)
 |    |    |-- title: string (nullable = true)
 |    |    |-- unit_price: double (nullable = true)
 |-- timestamp: string (nullable = true)
 |-- type: string (nullable = true)

+--------------+---------------+--------------------------------------------------------------------------------------------------+-------------------+-----+
|country       |invoice_no     |items                                                                                             |timestamp          |type |
+--------------+---------------+--------------------------------------------------------------------------------------------------+-------------------+-----+
|United Kingdom|154132541653705|[{21485, 6, RETROSPOT HEART HOT WATER BOTTLE, 4.95}, {23499, 2, SET 12 VINTAGE DOILY CHALK, 0.42}]|2020-09-18 10:55:23|ORDER|
+--------------+---------------+--------------------------------------------------------------------------------------------------+-------------------+-----+