how to run spark to read from json file and display content?

user_898042 0
Viewed 251

the spark_job.py file has this:

from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.sql.types import IntegerType, LongType, DecimalType,StructType, StructField, StringType
from pyspark.sql import Row
from pyspark.sql.functions import col
import pyspark.sql.functions as F
from pyspark.sql import Window

def readMyStream(rdd):
  if not rdd.isEmpty():
    df = spark.read.json(rdd)
    print('Started the Process')
    print('Selection of Columns')
    df = df.select('t1','t2','t3','timestamp').where(col("timestamp").isNotNull())
    df.show()

if __name__ == '__main__':
    sc = SparkContext.getOrCreate()
    spark = SparkSession(sc)
    ssc = StreamingContext(sc, 5)

    stream_data = ssc.textFileStream("jsondata.json")
    stream_data.foreachRDD( lambda rdd: readMyStream(rdd) )
    ssc.start()
    ssc.stop()

the jsondata.json file has this:

[{"timestamp": "1571053218000","t1": "55.23","t2": "10","t3": "ON"},

{"timestamp": "1571053278000","t1": "63.23","t2": "11","t3": "OFF"},

{"timestamp": "1571053338000","t1": "73.23","t2": "12","t3": "ON"},

{"timestamp": "1571053398000","t1": "83.23","t2": "13","t3": "ON"}]

running:

python spark_job.py

just gives me this:

Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
PS C:\Users\Admin\Desktop\madi_kafka> SUCCESS: The process with PID 10272 (child process of PID 2544) has been terminated.
SUCCESS: The process with PID 2544 (child process of PID 10652) has been terminated.
SUCCESS: The process with PID 10652 (child process of PID 4516) has been terminated.
apache-spark
2 Answers

show function can help you,I think this code example could help you to solve your problem:

val data = session.sqlContext.read.format("json").load("data/input.json")
val first = data.show()

Most of time spark could find your the schema of your data implicitly.

user_52163090

I use below code in scala,I think it could be helpful :

import session.implicits._
case class TClass(timestamp:String,t1:String,t2:String,t3:String)
val jsonData= session.read.option("inferSchema","true").option("multiline","true").option("header","true").json("data/jsondata.json").as[TClass]
jsonData.printSchema()
jsonData.show()
print("Started the Process")
print("Selection of Columns")
val df = jsonData.select("timestamp","t1","t2","t3").where(col("timestamp") isNotNull)
df.show()

and get this out:

+-------------+-----+---+---+
|    timestamp|   t1| t2| t3|
+-------------+-----+---+---+
|1571053218000|55.23| 10| ON|
|1571053278000|63.23| 11|OFF|
|1571053338000|73.23| 12| ON|
|1571053398000|83.23| 13| ON|
+-------------+-----+---+---+

I wish it could be helpful for you.

user_52163090
Related