您好,登录后才能下订单哦!
这篇“Spark Streaming编程初级源码分析”文章的知识点大部分人都不太理解,所以小编给大家总结了以下内容,内容详细,步骤清晰,具有一定的借鉴价值,希望大家阅读完这篇文章能有所收获,下面我们一起来看看这篇“Spark Streaming编程初级源码分析”文章吧。
Linux:CentOS7.5
Spark: spark-3.0.0-bin-hadoop3.2
Flume:Flume-
1.9.0
IDE:IntelliJ IDEA2020.2.3
Flume是Cloudera提供的一个分布式、可靠、可用的系统,它能够将不同数据源的海量日志数据进行高效收集、聚合、移动,最后存储到一个中心化数据存储系统中。Flume 的核心是把数据从数据源收集过来,再送到目的地。
或者也可以直接到本教程官网的“下载专区”中的“软件”目录中下载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
Avro可以发送一个给定的文件给Flume,Avro 源使用AVRO RPC机制。请对Flume的相关配置文件进行设置,从而可以实现如下功能:在一个终端中新建一个文件helloworld.txt(里面包含一行文本“Hello World”),在另外一个终端中启动Flume以后,可以把helloworld.txt中的文本内容显示出来。
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
请对Flume的相关配置文件进行设置,从而可以实现如下功能:在一个Linux终端(这里称为“Flume终端”)中,启动Flume,在另一个终端(这里称为“Telnet终端”)中,输入命令“telnet localhost 44444”,然后,在Telnet终端中输入任何字符,让这些字符可以顺利地在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
会话窗口成功得到数据
Flume是非常流行的日志采集系统,可以作为Spark Streaming的高级数据源。请把Flume Source设置为netcat类型,从终端上不断给Flume Source发送各种消息,Flume把消息汇集到Sink,这里把Sink类型设置为avro,由Sink把消息推送给Spark Streaming,由自己编写的Spark Streaming应用程序对消息进行处理。
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() } }
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) } } }
以上就是关于“Spark Streaming编程初级源码分析”这篇文章的内容,相信大家都有了一定的了解,希望小编分享的内容对大家有帮助,若想了解更多相关的知识内容,请关注亿速云行业资讯频道。
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。