基于Structured Streaming完成业务数据实时分析

妖狐艹你老母 2022-12-11 13:55 168阅读 0赞

### 目录 ###

*   *  一、熟悉业务流程
     *  二、启动服务
     *  三、启动业务程序
     *   *   *  （1）通过脚本启动flume
             *  （2）测试模拟程序服务
     *  四、Structured Streaming处理kafka的数据
     *  五、将处理的数据结果写入到MySQL数据库中

## 一、熟悉业务流程 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]  
项目数据流程：数据源——Flume采集——Kafka——Structured Streaming数据分析——MySQL数据库  
前半部分的数据采集Flume与Kafka的集成已经完工，详细可见这篇博客：  
[https://blog.csdn.net/weixin\_45366499/article/details/107154169][https_blog.csdn.net_weixin_45366499_article_details_107154169]

接下来的任务是：

*  数据源kafka-业务数据流\[weblogs\]
 *  结构化流处理kafka的数据
 *  处理结果写入MySQL数据库
 *  输出：titleName，count
 *  过滤前20名

## 二、启动服务 ##

在开始运行之前要启动所有的服务  
**启动Hadoop（node1）**

start-all.sh

**启动Zookeeper（node1，node2，node3）**

zkServer.sh start

**启动HBase（node1）**

start-hbase.sh

**启动Kafka**  
在/opt/Hadoop/kafka目录下启动  
在前台启动kafka，注意查看打印在桌面的日志，有无报错信息

bin/kafka-server-start.sh config/server.properties

如果没有报错信息，启动正常，那么就可以在后台启动了

bin/kafka-server-start.sh -daemon config/server.properties

jps查看进程  
node1:

3297 Jps
    20147 Application
    19203 ResourceManager
    20051 HRegionServer
    3268 Kafka
    18758 NameNode
    18855 DataNode
    19497 NodeManager
    19788 HMaster
    19053 SecondaryNameNode
    23343 ConsoleConsumer
    15167 QuorumPeerMain

node2:

19923 NodeManager
    19811 DataNode
    3348 Application
    20118 HRegionServer
    2699 Kafka
    10156 Jps
    16381 QuorumPeerMain
    20350 HMaster

node3:

18352 HRegionServer
    6547 Jps
    14907 QuorumPeerMain
    32205 Kafka
    18045 DataNode
    18157 NodeManager
    32319 Application

说明相关服务全部启动起来了。

## 三、启动业务程序 ##

#### （1）通过脚本启动flume ####

分别在每台机的在`/opt/Hadoop/flume`目录下执行：

./flume-caizhengjie-start.sh

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]  
出现这种情况，不报错，即为启动成功

#### （2）测试模拟程序服务 ####

在node2和node3`/opt/datas`目录下，执行命令：

./weblog-shell.sh

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]  
从结果看运行成功，数据的乱码被转换成了中文，也可以去通过查看weblog-flume.log日志文件，即数据被写入到这里面。

## 四、Structured Streaming处理kafka的数据 ##

读取kafka数据源之后，Structured Streaming要开始处理数据，运行下面程序：

package com.spark.test
    
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.sql.streaming.ProcessingTime
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/9/29 * @time : 4:49 下午 */
    object StructuredStreamingKafka { 
    
        case class Weblog(datatime:String,
                          userid:String,
                          searchname:String,
                          retorder:String,
                          cliorder:String,
                          cliurl:String)
    
        def main(args: Array[String]): Unit = { 
            // 创建Spark Session对象
            val spark = SparkSession
                    .builder
                    .master("local[2]")
                    .appName("streaming")
                    .getOrCreate()
    
            // 读取kafka stream流
            val df = spark
                    .readStream
                    .format("kafka")
                    .option("kafka.bootstrap.servers", "node1:9092")
                    .option("subscribe", "weblogs")
                    .load() //返回的是dataframe格式
    
            import spark.implicits._
            // 先将dataframe准换成dataset，在对数据进行处理
            val lines = df.selectExpr("CAST(value AS STRING)").as[String]
            val weblog =  lines.map(_.split(","))
                    .map(x => Weblog(x(0),x(1),x(2),x(3),x(4),x(5)))
            val titleCounts = weblog
                    .groupBy("searchname")
                    .count()
                    .toDF("titleName","count")
    
            val query = titleCounts.writeStream
                    .outputMode("update")
                    .format("console")
                    .trigger(ProcessingTime("5 seconds"))
                    .start()
                    
            query.awaitTermination()
    
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

## 五、将处理的数据结果写入到MySQL数据库中 ##

在pom.xml文件中添加如下配置：

<dependency>
          <groupId>mysql</groupId>
          <artifactId>mysql-connector-java</artifactId>
          <version>5.1.38</version>
    </dependency>

StructuredStreamingKafka程序：

package com.spark.test
    
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.sql.streaming.ProcessingTime
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/9/29 * @time : 4:49 下午 */
    object StructuredStreamingKafka { 
    
        case class Weblog(datatime:String,
                          userid:String,
                          searchname:String,
                          retorder:String,
                          cliorder:String,
                          cliurl:String)
    
        def main(args: Array[String]): Unit = { 
            // 创建Spark Session对象
            val spark = SparkSession
                    .builder
                    .master("local[2]")
                    .appName("streaming")
                    .getOrCreate()
    
            // 读取kafka stream流
            val df = spark
                    .readStream
                    .format("kafka")
                    .option("kafka.bootstrap.servers", "node1:9092")
                    .option("subscribe", "weblogs")
                    .load() //返回的是dataframe格式
    
            import spark.implicits._
            // 先将dataframe准换成dataset，在对数据进行处理
            val lines = df.selectExpr("CAST(value AS STRING)").as[String]
            val weblog =  lines.map(_.split(","))
                    .map(x => Weblog(x(0),x(1),x(2),x(3),x(4),x(5)))
            val titleCounts = weblog
                    .groupBy("searchname")
                    .count()
                    .toDF("titleName","count")
    
    
            val url = "jdbc:mysql://node1:3306/test"
            val username = "root"
            val password = "199911"
    
            // 输出到mysql数据库
            val writer = new JDBCSink(url,username,password)
            val query = titleCounts.writeStream
                    .foreach(writer)
                    .outputMode("update")
    // .format("console")
                    .trigger(ProcessingTime("5 seconds"))
                    .start()
    
            query.awaitTermination()
    
        }
    }

编写JDBCSink程序：

package com.spark.test
    
    import java.sql.{ Connection, DriverManager, ResultSet, Statement}
    
    import org.apache.spark.sql.{ ForeachWriter, Row}
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/10/2 * @time : 2:46 下午 */
    class JDBCSink(url:String,username:String ,password:String) extends ForeachWriter[Row]{ 
    
        var connection :Connection = _
        var statement : Statement = _
        var resultSet : ResultSet = _
    
        override def open(partitionId: Long, epochId: Long): Boolean = { 
            Class.forName("com.mysql.jdbc.Driver")
            connection = DriverManager.getConnection(url,username,password)
            statement = connection.createStatement()
            return true
        }
    
        override def process(value: Row): Unit = { 
    
            val titleName = value.getAs[String]("titleName")
            val count = value.getAs[Long]("count")
    
            // 查询语句
            val querySql = "select 1 from webCount where titleName = '"+titleName+"'"
    
            // 修改语句
            val updateSql = "update webCount set count = "+count+" where titleName = '"+titleName+"'"
    
            // 增加语句
            val insertSql = "insert into webCount(titleName,count) values('"+titleName+"',"+count+")"
    
            // 查询titleName，如果表中没有titleName，就增加titleName和count，如果表中有titleName，就修改count
            resultSet = statement.executeQuery(querySql)
            if(resultSet.next()){ 
                statement.executeUpdate(updateSql)
            }else{ 
                try{ 
                    statement.execute(insertSql)
                }
                catch { 
                    case ex:SQLException =>{ 
                        println("IO Exception")
                    }
                }
            }
        }
    
        override def close(errorOrNull: Throwable): Unit = { 
            statement.close()
            connection.close()
        }
    }

查看mysql数据库：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/8fb253618a7f4dd387fd48c22c7daeab.png
[https_blog.csdn.net_weixin_45366499_article_details_107154169]: https://blog.csdn.net/weixin_45366499/article/details/107154169
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]: /images/20221123/72e0136e1b4447e4a754b6cfbcb11c1a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]: /images/20221123/18b578b05a8048129bf7bdbcb1982f18.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]: /images/20221123/f2bfabd3fa07484ba18872716943f1a6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/04ac21813aee4e419e556a1ab9918e30.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/36b04abf12394ee480d958a846cf05b7.png