从kafka流pyspark中的嵌套json获取数据 [英] Get data from nested json in kafka stream pyspark

问题描述

我有一个kafka生产者，以以下格式发送大量数据

I have a kafka producer sending large amounts of data in the format of

{
  '1000': 
    {
       '3': 
        {
           'seq': '1', 
           'state': '2', 
           'CMD': 'XOR' 
        }
    },
 '1001': 
    {
       '5': 
        {
           'seq': '2', 
           'state': '2', 
           'CMD': 'OR' 
        }
    },
 '1003': 
    {
       '5': 
        {
           'seq': '3', 
           'state': '4', 
           'CMD': 'XOR' 
        }
    }
}

....我想要的数据在最后一个循环中: {'seq':'1'，'state':'2'，'CMD':'XOR'} 和上面循环中的键("1000"和"3")是可变的.请注意，以上值仅作为示例.原始数据集非常庞大，其中包含许多可变键.仅最后循环中的键 {'seq'，'state'，'CMD'} 是恒定的.

.... the data I want is in the final loop: {'seq': '1', 'state': '2', 'CMD': 'XOR'} and the keys in the loops above('1000' and '3') are variable. Please note that the above values are only for example. the original dataset is huge with lots of variable keys. only the keys in the final loop{'seq', 'state', 'CMD'} are constant.

我尝试使用通用格式读取数据，但是由于上面的循环具有可变键，因此我获取的数据不正确，我不确定如何定义模式来解析这种数据格式.

I have tried using the generic formats to read the data but am getting incorrect data since the loops above have variable keys and I am not sure how to define the schema to parse this format of data.

我要实现的输出是以下格式的数据框

The output I am trying to achieve is a dataframe of the format

seq    state     CMD
----------------------
 1       2       XOR
 2       2        OR
 3       4       XOR

推荐答案

这可能对您有用-使用 explode() 和 getItem() 如下-

This can be a working soluting for you - use explode() and getItem() as below-

a_json={
  '1000': 
    {
       '3': 
        {
           'seq': '1', 
           'state': '2', 
           'CMD': 'XOR' 
        }
    }
}
df = spark.createDataFrame([(a_json)])
df.show(truncate=False)

+-----------------------------------------+
|1000                                     |
+-----------------------------------------+
|[3 -> [CMD -> XOR, state -> 2, seq -> 1]]|
+-----------------------------------------+

此处的逻辑

df = df.select("*", F.explode("1000").alias("x", "y"))
df = df.withColumn("seq", df.y.getItem("seq")).withColumn("state", df.y.getItem("state")).withColumn("CMD", df.y.getItem("CMD"))
df.show(truncate=False)


 +-----------------------------------------+---+----------------------------------+---+-----+---+
|1000                                     |x  |y                                 |seq|state|CMD|
+-----------------------------------------+---+----------------------------------+---+-----+---+
|[3 -> [CMD -> XOR, state -> 2, seq -> 1]]|3  |[CMD -> XOR, state -> 2, seq -> 1]|1  |2    |XOR|
+-----------------------------------------+---+----------------------------------+---+-----+---+

根据进一步的输入更新代码

#Assuming that all the json columns are in a single column, hence making it an array column first.
df = df.withColumn("array_col", F.array("1000", "1001", "1003"))
#Then explode and getItem
df = df.withColumn("explod_col", F.explode("array_col"))
df = df.select("*", F.explode("explod_col").alias("x", "y"))
df_final = df.withColumn("seq", df.y.getItem("seq")).withColumn("state", df.y.getItem("state")).withColumn("CMD", df.y.getItem("CMD"))
df_final.select("seq","state","CMD").show()
|seq|state|CMD|
+---+-----+---+
|  1|    2|XOR|
|  2|    2| OR|
|  3|    4|XOR|
+---+-----+---+

这篇关于从kafka流pyspark中的嵌套json获取数据的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！