spark\u job.py文件包含以下内容：

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

jsondata.json文件包含以下内容：

[{"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"}]

跑步：

python spark_job.py

给我这个：

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.