Spark Streaming编程初级实践详解

更新时间：2023年04月20日 09:31:38 作者：WHYBIGDATA

这篇文章主要为大家介绍了Spark Streaming编程初级实践详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪

写在前面

Linux：CentOS7.5
Spark： spark-3.0.0-bin-hadoop3.2
Flume：Flume-1.9.0
IDE：IntelliJ IDEA2020.2.3

1. 安装Flume

Flume是Cloudera提供的一个分布式、可靠、可用的系统，它能够将不同数据源的海量日志数据进行高效收集、聚合、移动，最后存储到一个中心化数据存储系统中。Flume 的核心是把数据从数据源收集过来，再送到目的地。请到Flume官网下载Flume1.7.0安装文件，下载地址如下：

www.apache.org/dyn/closer.…

或者也可以直接到本教程官网的“下载专区”中的“软件”目录中下载apache-flume-1.7.0-bin.tar.gz。

下载后，把Flume1.7.0安装到Linux系统的“/usr/local/flume”目录下，具体安装和使用方法可以参考教程官网的“实验指南”栏目中的“日志采集工具Flume的安装与使用方法。

安装命令

tar -zxvf apache-flume-1.9.0-bin.tar.gz -C /export/server/
mv apache-flume-1.9.0-bin/ flume-1.9.0
sudo vi /etc/profile
export FLUME_HOME=/usr/local/flume
export PATH=$PATH:$FLUME_HOME/bin
source /etc/profile
mv flume-env.sh.template flume-env.sh

查看版本号

bin/flume-ng version

2.使用Avro数据源测试Flume

题目描述

Avro可以发送一个给定的文件给Flume，Avro 源使用AVRO RPC机制。请对Flume的相关配置文件进行设置，从而可以实现如下功能：在一个终端中新建一个文件helloworld.txt（里面包含一行文本“Hello World”），在另外一个终端中启动Flume以后，可以把helloworld.txt中的文本内容显示出来。

Flume配置文件

al.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.channels= c1
a1.sources.r1.bind = 0.0.0.0
al.sources.r1.port = 4141
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
al.channels.c1.capacity = 1000
a1.channels.c1.transaction = 100
al.sources.r1.channels = c1
a1.sinks.k1.channel=c1

执行命令

先进入到Flume安装目录，执行以下第一行命令；
开始新的一个会话窗口，执行第二行命令写入数据到指定的文件中
查看上一步骤中指定的文件内容

./bin/flume-ng agent -c . -f ./conf/avro.conf -n a1 -Dflume.root.logger=INFO,console
echo 'hello,world' >> ./log.00
bin/flume-ng avro-client --conf conf -H localhost -p 4141 -F ./log.00

执行结果如下

3. 使用netcat数据源测试Flume

题目描述

请对Flume的相关配置文件进行设置，从而可以实现如下功能：在一个Linux终端（这里称为“Flume终端”）中，启动Flume，在另一个终端（这里称为“Telnet终端”）中，输入命令“telnet localhost 44444”，然后，在Telnet终端中输入任何字符，让这些字符可以顺利地在Flume终端中显示出来。

编写Flume配置文件

al.sources = r1
a1.sinks = k1
a1.channels = c1
al.sources.r1.type = netcat
al.sources.r1.channels = c1
a1.sources.r1.bind = localhost
al.sources.r1.port = 44444
a1.sinks.k1.type = logger
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
al.channels.c1.transaction = 100
al.sources.r1.channels = c1
a1.sinks.k1.channel = c1

执行以下命令

./bin/flume-ng agent -c . -f ./netcatExample.conf -n a1 -Dflume.root.logger=INFO,console
telnet localhost 44444

会话窗口成功得到数据

4. 使用Flume作为Spark Streaming数据源

题目描述

Flume是非常流行的日志采集系统，可以作为Spark Streaming的高级数据源。请把Flume Source设置为netcat类型，从终端上不断给Flume Source发送各种消息，Flume把消息汇集到Sink，这里把Sink类型设置为avro，由Sink把消息推送给Spark Streaming，由自己编写的Spark Streaming应用程序对消息进行处理。

编写Flume配置文件

al.sources = r1
a1.sinks = k1
a1.channels =  c1
al.sources.r1.type = netcat
al.sources.r1.bind = localhost
a1.sources.r1.port = 33333
a1.sinks.k1.type = avro
al.sinks.k1.hostname = localhost
a1.sinks.k1.port = 44444
a1.channels.c1.type = memory
al.channels.c1.capacity = 1000000
a1.channels.c1.transactionCapacity = 1000000
al.sources.r1.channels = c1
a1.sinks.k1.channel = c1

主程序代码

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming._
import org.apache.spark.streaming.Milliseconds
import org.apache.spark.streaming.flume._
import org.apache.spark.util.IntParam
object FlumeEventCount {
    def main(args: Array[String]): Unit = {
        if (args.length < 2) {
            System.err.println( "Usage: FlumeEventCount <host> <port>")
            System.exit(1)
        }
        StreamingExamples.setStreamingLogLevels()
        val Array(host, IntParam(port)) = args
        val batchInterval = Milliseconds(2000)
        val sc = new SparkConf()
          .setAppName("FlumeEventCount")
//          .setMaster("local[2]")
        val ssc = new StreamingContext(sc, batchInterval)
        val stream = FlumeUtils.createStream(ssc, host, port, StorageLevel.MEMORY_ONLY_SER_2)
        stream.count().map(cnt => "Received " + cnt + " flume events." ).print()
        ssc.start()
        ssc.awaitTermination()
    }
}

执行结果1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.internal.Logging
object StreamingExamples extends Logging {
    def setStreamingLogLevels(): Unit = {
        val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
        if (!log4jInitialized) {
            logInfo("Setting log level to [WARN] for streaming example." + " To override add a custom log4j.properties to the classpath.")
            Logger.getRootLogger.setLevel(Level.WARN)
        }
    }
}

执行结果2

以上就是Spark Streaming编程初级实践详解的详细内容，更多关于Spark Streaming编程的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

SpringSecurity集成图片验证码的详细过程
SpringSecurity是通过过滤器链来完成的,接下来的验证码,可以尝试创建一个过滤器放到Security的过滤器链中,在自定义的过滤器中比较验证码,本文通过实例代码介绍SpringSecurity集成图片验证码的详细过程,感兴趣的朋友一起看看吧
2023-12-12
Spring question问题小结
在AppConfig配置类中,通过@Bean注解创建了Service和Controller的实例,Spring会自动将这些实例纳入容器的管理,并处理它们之间的依赖关系,本文给大家介绍Spring question问题小结,感兴趣的朋友跟随小编一起看看吧
2023-10-10
mybatis-plus实现自定义SQL、多表查询与多表分页查询语句实例
mybatisplus是个很好用的插件,相信小伙伴们都知道,下面这篇文章主要给大家介绍了关于mybatis-plus实现自定义SQL、多表查询与多表分页查询语句的相关资料,需要的朋友可以参考下
2022-09-09
Java Proxy机制详细解读
这篇文章主要介绍了Java Proxy机制详细解读，还是非常不错的，这里分享给大家，需要的朋友可以参考下。
2017-10-10
SpringBoot实现邮件发送功能的姿势分享
我们在日常开发中,经常会碰到email邮件发送的场景,如发送验证码,向客户发送邮件等等,这篇文章主要给大家介绍了关于SpringBoot实现邮件发送的相关资料,需要的朋友可以参考下
2021-08-08
Java案例之随机验证码功能实现实例
本篇文章主要介绍了Java案例之随机验证码功能实现实例，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-06-06
Java中的ThreadLocal功能演示示例
这篇文章主要介绍了Java中的ThreadLocal功能演示示例，帮助大家更好的理解和使用Java，感兴趣的朋友可以了解下
2021-02-02
mybatis映射和实际类型不一致的问题
这篇文章主要介绍了mybatis映射和实际类型不一致的问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-11-11
如何在SpringBoot+Freemarker中获取项目根目录
这篇文章主要介绍了如何在SpringBoot+Freemarker中获取项目根目录的操作，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-10-10
MVC框架自定义实现过程
这篇文章主要介绍了MVC框架自定义实现过程，通过搞懂什么是MVC以及MVC的运行原理然后再去理解其演绎过程，本文附含了详细的思维导图帮助大家更好的理解MVC
2021-08-08