pyspark - How to convert spark streaming nested json coming on kafka to flat dataframe? -

-

need on first attempt parse json coming on kafka spark structured streaming.

i struggling convert incoming json , covert flat dataframe further processing.

my input json 

[     { "siteid": "30:47:47:be:16:8f", "sitedata":          [             { "dataseries": "trend-255", "values":                  [                     {"ts": 1502715600, "value": 35.74 },                     {"ts": 1502715660, "value": 35.65 },                     {"ts": 1502715720, "value": 35.58 },                     {"ts": 1502715780, "value": 35.55 }                 ]             },             { "dataseries": "trend-256", "values":                 [                     {"ts": 1502715840, "value": 18.45 },                     {"ts": 1502715900, "value": 18.35 },                     {"ts": 1502715960, "value": 18.32 }                 ]             }         ]     },     { "siteid": "30:47:47:be:16:ff", "sitedata":          [             { "dataseries": "trend-255", "values":                  [                     {"ts": 1502715600, "value": 35.74 },                     {"ts": 1502715660, "value": 35.65 },                     {"ts": 1502715720, "value": 35.58 },                     {"ts": 1502715780, "value": 35.55 }                 ]             },             { "dataseries": "trend-256", "values":                 [                     {"ts": 1502715840, "value": 18.45 },                     {"ts": 1502715900, "value": 18.35 },                     {"ts": 1502715960, "value": 18.32 }                 ]             }         ]     } ]

spark schema is

data1_spark_schema = arraytype( structtype([   structfield("siteid", stringtype(), false),   structfield("sitedata", arraytype(structtype([     structfield("dataseries", stringtype(), false),     structfield("values", arraytype(structtype([       structfield("ts", integertype(), false),       structfield("value", stringtype(), false)     ]), false), false)   ]), false), false) ]), false )

my simple code is:

from pyspark.sql import sparksession pyspark.sql.functions import *  config.general import kafka_instance config.general import topic schemas.schema import data1_spark_schema  spark = sparksession \             .builder \             .appname("structured_bms_feed") \             .getorcreate()  stream = spark \         .readstream \         .format("kafka") \         .option("kafka.bootstrap.servers", kafka_instance) \         .option("subscribe", topic) \         .option("startingoffsets", "latest") \         .option("max.poll.records", 100) \         .option("failondataloss", false) \         .load()  stream_records = stream.selectexpr("cast(key string)", "cast(value string) bms_data1") \                        .select(from_json("bms_data1", data1_spark_schema).alias("bms_data1"))  sites = stream_records.select(explode("bms_data1").alias("site")) \                       .select("site.*")  sites.printschema()  stream_debug = sites.writestream \                              .outputmode("append") \                              .format("console") \                              .option("numrows", 20) \                              .option("truncate", false) \                              .start()   stream_debug.awaittermination()

when run code schema printing this:

root  |-- siteid: string (nullable = false)  |-- sitedata: array (nullable = false)  |    |-- element: struct (containsnull = false)  |    |    |-- dataseries: string (nullable = false)  |    |    |-- values: array (nullable = false)  |    |    |    |-- element: struct (containsnull = false)  |    |    |    |    |-- ts: integer (nullable = false)  |    |    |    |    |-- value: string (nullable = false)

is possible have schema in way fields in flat dataframe instead of nested json. every ts , value should give me 1 row parent dataseries , site id.

answering own question. managed flatten using following lines:

sites_flat = stream_records.select(explode("bms_data1").alias("site")) \                            .select("site.siteid", explode("site.sitedata").alias("sitedata")) \                            .select("siteid", "sitedata.dataseries", explode("sitedata.values").alias("values")) \                            .select("siteid", "dataseries", "values.*")

Comments

Post a Comment

Popular posts from this blog

ios - MKAnnotationView layer is not of expected type: MKLayer -

-
Image
so code works fine logger riddled message. there way rid of or suppress it? postannotation.swift class postannotation: mkpointannotation { //mark: properties let post: post //mark: initialization init(post: post) { self.post = post super.init() self.coordinate = cllocationcoordinate2d(latitude: post.latitude, longitude: post.longitude) self.title = post.title self.subtitle = post.timestring() } } adding annotation let annotation = postannotation(post: post) self.map.addannotation(annotation) func mapview func mapview(_ mapview: mkmapview, viewfor annotation: mkannotation) -> mkannotationview? { if annotation mkuserlocation { return nil } var annotationview = mapview.dequeuereusableannotationview(withidentifier: "pin") as? mkpinannotationview if annotationview == nil { annotationview = mkpinannotationview(annotation: annotation, reuseidentifier: "pin")
Read more

ZeroMQ on Windows, with Qt Creator -

-
i have installed zmq link given below on windows 7. http://zeromq.org/distro:microsoft-windows 1) have c++ project in qt-creator , i'm using zmq c++ binding (zhelpers.hpp etc). 2) have added .lib installed zmq folder qtcreator 3) while compiling, compiler looks _imp__zmq_verion, in .dll files present in .lib have __imp_zmq_version. 4) causing undefined reference zmq_version. anyone else faced problem , has solution ?
Read more

unity3d - Unity SceneManager.LoadScene quits application -

-
i have 2 different scenes 1 named 'main' , other 'game' both added scenes in load. when use scenemanager.loadscene("game"); works no problem when tried return main scene code if (input.getkeydown (keycode.escape)) { scenemanager.loadscene("main"); } strangely game quits. thanks. sorry interruption. somewhere on other script found code. if (input.getkeydown (keycode.escape)) { application.quit (); } that's bad, know...
Read more