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
#兩個參數：1、sc參數?2、采樣時間間隔（秒）
ssc?=StreamingContext（sc1）
#在ubuntu環境下數據源路勁
ds1?=ssc.textFileStream（“/home/zhuang/138/input/test“）
#把所有數據劃分為[[][]]格式
ds3?=?ds1.map（lambda?line:line.split（“\t“））
def?func（rdd）:
????if?not?rdd.isEmpty（）:
????????#記得轉碼很重要
????????url?=?“jdbc:mysql://ip地址:3306/pyspark?user=root&password=zhuang&characterEncoding=UTF-8“
????????#構建表結構
????????schema?=?StructType（[StructField（“日志級別“?StringType（）?True）?StructField（“函數名“?StringType（）?True）
?????????????????????????????StructField（“日志內容“?StringType（）?True）]）
????????#對[[][][]]數據轉換成[[[][][]][][][]]因為todf數據是數據格式傳值
????????rdd.map（lambda?x:tuple（x））.toDF（schema）.registerTempTable（“test_person1“）
????????df1?=?spark.sql（“select?*?from?test_person1?where?‘日志級別‘!=‘[info]‘“）
????????#?df2?=?spark
????????df1.show（）
????????#寫入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（）

?屬性????????????大小?????日期????時間???名稱
-----------?---------??----------?-----??----
?????文件?????????501??2019-05-04?11:03??20180103.log
?????文件?????????501??2019-05-04?11:03??20180104.log
?????文件?????1007502??2019-03-12?14:38??mysql-connector-java-5.1.47.jar
?????文件????????1514??2019-05-30?08:58??test02.py