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)
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.