How complex and nested json dataset works with pyspark

Viewed 544

I have a very complex data and was handled in scala in databricks. I want to convert that scala to python and that should work with the data given which is in JSON

Scala code:

import org.apache.spark.sql.types._                        
import org.apache.spark.sql.functions._                     

val jsonSchema = new StructType()
        .add("battery_level", LongType)
        .add("c02_level", LongType)
        .add("cca3",StringType)
        .add("cn", StringType)
        .add("device_id", LongType)
        .add("device_type", StringType)
        .add("signal", LongType)
        .add("ip", StringType)
        .add("temp", LongType)
        .add("timestamp", TimestampType)

// define a case class

case class DeviceData (id: Int, device: String)

// create some sample data

val eventsDS = Seq (

(0, """{"device_id": 0, "device_type": "sensor-ipad", "ip": "68.161.225.1", "cca3": "USA", "cn": "United States", "temp": 25, "signal": 23, "battery_level": 8, "c02_level": 917, "timestamp" :1475600496 }"""),

 (1, """{"device_id": 1, "device_type": "sensor-igauge", "ip": "213.161.254.1", "cca3": "NOR", "cn": "Norway", "temp": 30, "signal": 18, "battery_level": 6, "c02_level": 1413, "timestamp" :1475600498 }"""),

 (2, """{"device_id": 2, "device_type": "sensor-ipad", "ip": "88.36.5.1", "cca3": "ITA", "cn": "Italy", "temp": 18, "signal": 25, "battery_level": 5, "c02_level": 1372, "timestamp" :1475600500 }"""),

 (3, """{"device_id": 3, "device_type": "sensor-inest", "ip": "66.39.173.154", "cca3": "USA", "cn": "United States", "temp": 47, "signal": 12, "battery_level": 1, "c02_level": 1447, "timestamp" :1475600502 }""")).toDF("id", "device").as[DeviceData]

display(eventsDS)

Click here to see the output

Now i want to implement above code in pyspark. I have done someway but stuck in Seq as no Seq is in python. How to handle this sample data in pyspark?

Python code:

from pyspark.sql import SparkSession 
from pyspark.sql.types import *
from pyspark.sql.functions import *
from dataclasses import dataclass

scSpark = SparkSession.builder.appName("complex data types").getOrCreate()
#Creating JSON schema
jsonSchema = StructType([
  StructField("battery_level", LongType(), True), 
  StructField("c02_level", IntegerType(), True),
  StructField("c02_level", LongType(), True),
  StructField("cca3",StringType(), True),
  StructField("cn", StringType(), True),
  StructField("device_id", LongType(), True),
  StructField("device_type", StringType(), True),
  StructField("signal", LongType(), True),
  StructField("ip", StringType(), True),
  StructField("temp", LongType(), True),
  StructField("timestamp", TimestampType(), True),
])
#Create a Dataset from the above schema
@dataclass
class DeviceData(object):
  id: int
  device: str

Now i don't understand what to write next. I want the output like in the image i was provided.

My main motive is to https://docs.databricks.com/spark/latest/dataframes-datasets/complex-nested-data.html convert all scala into pyspark. That will solve my problem.

1 Answers

A scala Seq is most comparable to a python list:

eventsDS = (sql.createDataFrame(
            [(0, """{"device_id": 0, "device_type": "sensor-ipad", "ip": "68.161.225.1", "cca3": "USA", "cn": "United States", "temp": 25, "signal": 23, "battery_level": 8, "c02_level": 917, "timestamp" :1475600496 }"""),
             (1, """{"device_id": 1, "device_type": "sensor-igauge", "ip": "213.161.254.1", "cca3": "NOR", "cn": "Norway", "temp": 30, "signal": 18, "battery_level": 6, "c02_level": 1413, "timestamp" :1475600498 }"""),
             (2, """{"device_id": 2, "device_type": "sensor-ipad", "ip": "88.36.5.1", "cca3": "ITA", "cn": "Italy", "temp": 18, "signal": 25, "battery_level": 5, "c02_level": 1372, "timestamp" :1475600500 }"""),
             (3, """{"device_id": 3, "device_type": "sensor-inest", "ip": "66.39.173.154", "cca3": "USA", "cn": "United States", "temp": 47, "signal": 12, "battery_level": 1, "c02_level": 1447, "timestamp" :1475600502 }""")],
            ['id', 'device']))

eventsDS.show()                                                                                                                                                                                           

+---+--------------------+                                                      
| id|              device|
+---+--------------------+
|  0|{"device_id": 0, ...|
|  1|{"device_id": 1, ...|
|  2|{"device_id": 2, ...|
|  3|{"device_id": 3, ...|
+---+--------------------+

In pyspark SQL, the case class is not necessary.

eventsDS.printSchema()                                                                                                                                                                                    

root
 |-- id: long (nullable = true)
 |-- device: string (nullable = true)
Related