目录

第六章 Spark Streaming 实验手册

2020-04-19
5分钟阅读时长
spark

【版本】

当前版本号v20200424

版本修改说明
v20200424新增修改日志级别的步骤
v20200419初始化版本

实验6.1 ：运行网络版的WordCount

【实验名称】 实验6.1 ：运行网络版的WordCount

【实验目的】

  • 了解NetCat工具的使用。
  • 初步了解Streaming运行的方式。

【实验原理】

运行Spark自带的StreamingWordCount程序，

【实验环境】

  • 操作系统：Ubuntu 16.04 （确保机器cpu核数大于接收器的数量，或Local模式线程数大于接收器数量）
  • Spark：Spark2.x
  • Pyspark

【实验步骤】

一、基于netcat的聊天室

  1. 启动 NetCat 服务端，并在1234端口监听
nc -lk  1234
  1. 使用xshell 打开一个新的选项卡，连接虚拟机。启动NetCat客户端，并连接Netcat服务端
nc localhost  1234

注意：如果客户端和服务端不在同一台机器，localhost 可以换成实际IP。

  1. 在服务端输入以下字符串，并按回车，可以在客户端收到消息，并打印出来。这里注意替换学号为你个人学号。
hello 你的学号
  1. 在客户端输入字符串，并按回车，可以在服务端收到消息，并打印出来。这里注意替换学号为你个人学号。
你好  你的学号

  1. 在 NetCat 客户端的选项卡使用ctrl+c终止客户端进程。

  2. 修改$SPARK_HOME/conf下的 log4j.properties 下的日志级别，把INFO修改为WARN。这一步主要是为了提升日志级别，减少 WARN 级别以下例如 INFO,DEBUG 级别的日志输出，对结果造成干扰。

#修改这一句
log4j.rootCategory=INFO, console

#修改为
log4j.rootCategory=WARN, console
  1. 启动 Spark NetworkWordCount 例子和服务端建立连接，并统计单词数量。
cd $SPARK_HOME
bin/run-example streaming.NetworkWordCount localhost 1234
  1. 在 NetCat 服务端输入以下字符串，并按回车，观察Streaming WordCount的输出，并截图。这里注意替换学号为你个人学号。
You jump I jump 你的学号

实验6.2 ：开发自己的StreamingWordCount，且支持统计历史数据

【实验名称】 实验6.2 ：开发自己的StreamingWordCount，且支持统计历史数据

【实验目的】

  • 了解NetCat工具的使用。
  • 掌握 Spark Streaming 的开发流程。

【实验原理】

开发自己的StreamingWordCount 程序，用于统计所有流数据的单词数量。

【实验环境】

  • 操作系统：Ubuntu 16.04 （确保机器cpu核数大于接收器的数量，或Local模式线程数大于接收器数量）
  • Spark：Spark2.x
  • Pyspark

【实验要求】

  1. 开发一个 Spark Streaming WordCount 程序，满足以下要求
  • （1）自己编写一个 Spark Streaming 的 WordCount 程序，保存为StreamingWordCount123.py。（请替换123为你学号后3位）
  • （2）程序需要使用到 transform() 和 updateStateByKey() 算子。
  • （3）程序需要对NetCat 服务端发送历史的单词出现的数量一起累计。
  • （4）使用 spark-submit 运行你的程序。
  1. 启动 NetCat 服务端，并在1234端口监听。
nc -lk  1234
  1. 使用 NetCat 服务端分别发送以下内容。这里注意替换学号为你个人学号。
Knock 你的学号

Knock Knock 你的学号

Knock Knock  Knock 你的学号

实验6.3：流式日志过滤与分析

【实验名称】 实验6.3：流式日志过滤与分析

【实验目的】

掌握Spark Steaming的应用

【实验原理】

【实验环境】

  • 操作系统：Ubuntu 16.04
  • Spark：Spark2.x
  • Pyspark

【实验资源】

请从以下链接下载日志文件。该日志包含日志级别、函数名、日志内容三个字段，字段之间以"\t"拆分。

https://pan.baidu.com/s/1Vh1w-70x_MpkV_Ycu_54PA#提取码38zq

【实验步骤】

  1. 启动 HDFS，创建 HDFS 目录/sst-123。请替换123为你的学号后3位。

  2. 使用 MySQL 的 root 用户创建数据库 spark。并创建表log123。请替换123为你的学号后3位。

用户名：root
密码：Mysql0668

create table `log123`(
lvl varchar(15) comment '等级',
method varchar(50) comment '方法',
content varchar(200) comment '内容'
);
  1. 为了让程序可以调用 MySQL 的JDBC驱动，把驱动文件放入$SPARK_HOME/jars。
cp /opt/hive/lib/mysql-connector-java-5.1.48.jar $SPARK_HOME/jars/
  1. 如果配置了Jupytor Notebook，要把这2个环境变量注释掉，并重启虚拟机。
#export PYSPARK_DRIVER_PYTHON=jupyter
#export PYSPARK_DRIVER_PYTHON_OPTS='notebook --ip=*'

  1. 启动 Spark Standalone 模式。

  2. 编写程序Log2DB.py，完成以下要求。

  • （1）采用StreamingContext.textFileStream()算子监听hdfs://node0:8020/sst-123目录。
  • （2）把 DStream 转换为 RDD。
  • （3）读入日志信息流，将 RDD 转为 DataFrame。
  • （4）DataFrame 注册为临时表。
  • （5）使用 SQL 过滤出级别为 error 或 warn 的日志。
  • （6）将过滤后的数据保存到 MySQL 的 spark.log123表（请替换123为你的学号后3位）。

提示代码(注意修改为你的学号后3位)：

from pyspark.streaming import StreamingContext  
from pyspark import SparkContext
from pyspark.sql import SparkSession

sc =SparkContext('spark://node0:7077','ss-123')
spark =SparkSession(sc)
ssc = StreamingContext(sc, 1)
ds1 = ssc.textFileStream(<补充代码>)
def save2DB(fileRDD):
    <补充代码>
    df2.write.jdbc('jdbc:mysql://localhost:3306/spark?useSSL=false&user=root&password=Mysql0668',table='log123',mode='append',properties={'driver':'com.mysql.jdbc.Driver'})

ds1.foreachRDD(lambda fileRDD:save2DB(fileRDD))
ssc.start();ssc.awaitTermination()
  1. 使用 spark-submit 运行Log2DB.py。
spark-submit --master spark://node0:7077 ./Log2DB.py

  1. 把2个日志文件依次上传hdfs://node0:8020/sst-123目录下。

  2. 观察 MySQL 的spark.log123表记录是否有更新。

【常见问题解答FAQ】

问题1：上传文件到 HDFS 遇到org.apache.hadoop.mapreduce.lib.input.InvalidInputException:Input path does not exist: hdfs://node0:8020/xxx.txt.COPYING之类的问题

答：可以先把文件上传到 Spark Streaming 非监听目录，例如 HDFS 根目录。再使用hdfs dfs -mv命令，把该文件移动到监听目录。因为直接上传文件会在 HDFS 上生成一个临时文件，后缀是.COPYING，Spark Streaming 程序监听到该临时文件时，文件可能会因为复制完成被删除。导致文件找不到出错。但是hdfs dfs -mv命令是一个事务性的操作，不会产生临时文件。

#spark #Streaming
Avatar
黄老师