Spark Steaming流式日志过滤与分析

这篇大概讲的是 spark steaming 监听 hdfs 的某个目录，当你在终端A使用 spark-submit 运行 Log2DB.py 文件后，在终端B依次上传那两个 .log 文件到 hdfs 这个目录后，那么 spark streaming 就会依据 Log2DB.py 文件按要求将清洗后的数据(日志文件中的 error 和 warn 级别的数据)保存在 MySQL 的 spark 数据库中的log032表上。
尤其需要注意的是先运行Log2DB.py后再将日志文件依次上传到监控目录中，这样才能起到监控的作用。

本网页数据源的下载链接

步骤：

1、启动hdfs

命令：

// 转到相应的工作目录
cd /home/syan/Hbase/hadoop/sbin

// 启动 HDFS
start-dfs.sh

// 查看开启Hadoop是否成功
jps

2、创建hdfs监控目录

// 创建 HDFS 目录/spark-exp6/sst-032。
hdfs dfs -mkdir -p /spark-exp6/sst-032

在ip:50070可以看到创建目录成功

3、使用 MySQL 的 root 用户创建数据库 spark，并创建表log032。

spark数据库的用户名和密码
用户名：spark
密码：spark

（1）打开Navicat连接MySQL数据库

（2）执行下列SQL语句创建spark数据库
注：IP地址是根据自身配置的

-- 创建spark数据库
create database spark;
-- 创建用户名密码为spark/spark的用户（）
create user 'spark'@'IP地址' identified by 'spark';
create user 'spark'@'%' identified by 'spark';
-- 授予权限
grant all on spark.* to 'spark'@'IP地址';
grant all on spark.* to 'spark'@'%';

可以看到spark数据库已经创建成功

（3）然后执行下列SQL语句在spark数据库中创建表log032

use spark;

-- 创建表log032
create table log032(lvl varchar(15) comment '等级',method varchar(50) comment '方法',content varchar(200) comment '内容');

（4）为了让程序可以调用 MySQL 的JDBC驱动，把驱动文件mysql-connector-java-5.1.48-bin.jar放入$SPARK_HOME/jars
用xftp从Windows本地传到Ubuntu里面去

注：如果配置了Jupytor Notebook，要在$SPARK_HOME/conf/spark-env.sh文件中把这2个环境变量注释掉，并重启虚拟机。
#export PYSPARK_DRIVER_PYTHON=jupyter
#export PYSPARK_DRIVER_PYTHON_OPTS='notebook --ip=*'

（5）启动 Spark Standalone 模式。
看网上很多资料是说启动spark集群就是启动Spark Standalone模式，即在$SPARK_HOME/sbin下运行start-spark.sh文件。

注：如果你不想每次都进入这个目录去执行这个文件，那你可以选择设置临时环境或永久性环境。设置方法请看Ubuntu中设置PATH变量值

// 进入相应的工作目录
cd /home/syan/Spark/spark/sbin
// 启动服务
start-spark.sh

jps查看是否启动成功

（6）在本地编写程序Log2DB.py，有以下的要求。

（1）采用StreamingContext.textFileStream()算子监听hdfs://syan:9000/spark-exp6/sst-032目录。
（2）把 DStream 转换为 RDD。
（3）读入日志信息流，将 RDD 转为 DataFrame。
（4）DataFrame 注册为临时表。
（5）使用 SQL 过滤出级别为 error 或 warn 的日志。
（6）将过滤后的数据保存到 MySQL 的 spark数据库中的log032表

代码块：

#导入相应的包
from pyspark.streaming import StreamingContext
from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql.types import *#创建SparkContext对象
sc = SparkContext('spark://syan:7077','Log2DB')#生成流计算上下文，每10秒批处理一次
spark =SparkSession(sc)
ssc = StreamingContext(sc, 10)'''
采用textFileStream()算子监控目录hdfs://syan:50070/spark/sst-032不知道端口号是多少的话可以用下面这条命令查看：
hdfs getconf -confKey fs.default.name
'''
ds1 = ssc.textFileStream("hdfs://syan:9000/spark-exp6/sst-032")#把所有数据划分为[[],[]]格式
ds2 = ds1.map(lambda line:line.split("\t"))
ds3 = ds2.map(lambda arr:[arr[0],arr[1],arr[2]])def save2DB(fileRDD):if not fileRDD.isEmpty():'''构建表结构:日志级别------->lvl函数名--------->method日志内容------->content'''schema = StructType([StructField("lvl", StringType(), True), StructField("method", StringType(), True),StructField("content", StringType(), True)])#对[[],[],[]]数据转换成[[[],[],[]],[],[],[]]因为todf数据是数据格式传值：DataFrame 注册为临时表。fileRDD.map(lambda x:tuple(x)).toDF(schema).registerTempTable("tmp_log032")#使用 SQL 过滤出级别为 error 或 warn 的日志。df1 = spark.sql("select * from tmp_log032 where lvl != '[info]'")# df2 = sparkdf1.show()#写入mysql，即将过滤后的数据保存到 MySQL 的 spark数据库中的log123表df1.write.jdbc('jdbc:mysql://192.168.45.128:3306/spark?useSSL=false&user=spark&password=spark&characterEncoding=UTF-8',table='log032',mode='append',properties={'driver':'com.mysql.jdbc.Driver'})#打印输出到控制台
ds3.pprint()# 输出采用foreachRDD()算子
ds3.foreachRDD(lambda fileRDD:save2DB(fileRDD))#执行程序
ssc.start()#等待任务执行结束
ssc.awaitTermination()

Log2DB.py代码完整截图：

（7）使用 spark-submit 运行Log2DB.py
将写好的Log2DB.py上传到虚拟机（自己选一个位置，这里的是我上传的路径而已：/home/syan/Spark/Demo/input）中

终端A：
然后先转到spark解压目录下的bin目录下($SPARK_HOME/bin)

cd $SPARK_HOME/bin

注：如果你不想每次都进入这个目录去执行这个文件，那你可以选择设置临时环境或永久性环境。设置方法请看Ubuntu中设置PATH变量值

用 spark-submit运行python文件

$ spark-submit --master spark://syan:7077 /home/syan/Spark/Demo/input/Log2DB.py

（8）将我们刚刚从网盘下载下来的20180101.log、20180102.log文件上传到虚拟机(/home/syan/course/spark_course/data)，再从虚拟机上传到hdfs中(/spark-exp6/sst-032)
1、把20180101.log、20180102.log文件一起上传到虚拟机(/home/syan/course/spark_course/data)

2、另外开一个终端：终端B分别把20180101.log、20180102.log(/home/syan/course/spark_course/data)从虚拟机上传到hdfs中(/spark-exp6/sst-032)

（1）终端B中上传20180101.log

终端B:

// 终端B中上传20180101.log
hdfs dfs -put /home/syan/course/spark_course/data/20180101.log /spark-exp6/sst-032

返回终端A查看打印结果：

再看看Ubuntu中的MySQL中的spark数据库的log032表有没有更新数据

（2）再回到终端B 中继续往hdfs中上传20180102.log文件

终端B:

// 终端B中上传20180102.log
hdfs dfs -put /home/syan/course/spark_course/data/20180102.log /spark-exp6/sst-032

返回终端A查看打印结果

再看看Ubuntu中的MySQL中的spark数据库的log032表有没有更新数据

在spark streaming输入源中的基本数据源是直接调用API的，而基本数据源的数据流中的文件流(textFileStream)中是通过监控文件系统目录下的变化，就比如要监控hdfs的目录，若是有新文件添加到监控目录，则将这个文件内容读入并作为数据流。但是它所监控的目录下的文件有下面三种要求：
（1）上传到监控目录中的文件要具有相同的格式。
（2）不一定是通过上传才能使能被监控，移动和重命名的方法都可以使得监控目录可以读取并将它写入到数据流中。
（3）但如果在已存在监控目录下的文件中写入或追加内容，这些追加的新数据不会被读取并写入到数据流中。