from?pyspark.shell?import?sc
from?pyspark.sql?import?Row
from?pyspark.sql?import?SparkSession
from?pyspark.sql.types?import?*
from?pyspark.streaming?import?StreamingContext
spark?=?SparkSession.builder.appName（“Streaming“）.getOrCreate（）
sc=spark.sparkContext
#兩個(gè)參數(shù)：1、sc參數(shù)?2、采樣時(shí)間間隔（秒）
ssc?=StreamingContext（sc1）
#在ubuntu環(huán)境下數(shù)據(jù)源路勁
ds1?=ssc.textFileStream（“/home/zhuang/138/input/test“）
#把所有數(shù)據(jù)劃分為[[][]]格式
ds3?=?ds1.map（lambda?line:line.split（“\t“））
def?func（rdd）:
????if?not?rdd.isEmpty（）:
????????#記得轉(zhuǎn)碼很重要
????????url?=?“jdbc:mysql://ip地址:3306/pyspark?user=root&password=zhuang&characterEncoding=UTF-8“
????????#構(gòu)建表結(jié)構(gòu)
????????schema?=?StructType（[StructField（“日志級(jí)別“?StringType（）?True）?StructField（“函數(shù)名“?StringType（）?True）
?????????????????????????????StructField（“日志內(nèi)容“?StringType（）?True）]）
????????#對(duì)[[][][]]數(shù)據(jù)轉(zhuǎn)換成[[[][][]][][][]]因?yàn)閠odf數(shù)據(jù)是數(shù)據(jù)格式傳值
????????rdd.map（lambda?x:tuple（x））.toDF（schema）.registerTempTable（“test_person1“）
????????df1?=?spark.sql（“select?*?from?test_person1?where?‘日志級(jí)別‘!=‘[info]‘“）
????????#?df2?=?spark
????????df1.show（）
????????#寫(xiě)入mysql
????????df1.write.jdbc（mode=“overwrite“url=urltable=“test_person1“?properties={“driver“:‘com.mysql.jdbc.Driver‘}）
????????df1.show（）
ds3.pprint（）
ds3.foreachRDD（func）
#?print（ds4.foreachRDD（func））
ssc.start（）;ssc.awaitTermination（）