本文主要是介绍Spark Streaming(三)—— 高级数据源Flume,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
文章目录
- 高级数据源Flume
- 1. Push方式
- 2. 基于Custom Sink的Pull模式
高级数据源Flume
Spark Streaming 是一个流式计算引擎,就需要对接外部数据源来对接、接收数据。每一个输入流DStream和一个Receiver对象相关联,这个Receiver从源中获取数据,并将数据存入内存中用于处理。Spark Streaming的基本数据源(文件流、RDD队列流、套接字流)上篇已经介绍过了,而Spark Streaming的高级数据流主要有Kafka,Flume,Kinesis,Twitter等。本文主要介绍Flume作为高级数据源的使用。
1. Push方式
Flume将数据推送给Spark Streaming。在这种方式下,Spark Streaming可以很方便的建立一个Receiver,起到一个Avro agent的作用。Flume可以将数据推送到该Receiver。
Flume配置文件:
#定义agent名, source、channel、sink的名称
a4.sources = r1
a4.channels = c1
a4.sinks = k1#具体定义source 采集该目录下的日志
a4.sources.r1.type = spooldir
a4.sources.r1.spoolDir = /root/training/logs#具体定义channel
a4.channels.c1.type = memory
a4.channels.c1.capacity = 10000
a4.channels.c1.transactionCapacity = 100#具体定义sink
a4.sinks = k1
a4.sinks.k1.type = avro
a4.sinks.k1.channel = c1
a4.sinks.k1.hostname = 192.168.15.131
a4.sinks.k1.port = 1234#组装source、channel、sink
a4.sources.r1.channels = c1
a4.sinks.k1.channel = c1
Spark Streaming Demo:
注意除了需要使用Flume的lib的jar包以外,还需要spark-streaming-flume_2.10-2.1.0.jar
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.flume.FlumeUtilsobject MyFlumeStream {def main(args: Array[String]): Unit = {System.setProperty("hadoop.home.dir", "G:\\bin\\hadoop-2.5.2")Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)val conf = new SparkConf().setAppName("MyFlumeStream").setMaster("local[2]")val ssc = new StreamingContext(conf,Seconds(3))//创建 flume event 从 flume中接收push来的数据 ---> 也是DStream//flume将数据push到了 ip 和 端口中 ip和端口在Flume配置文件中设置val flumeEventDstream = FlumeUtils.createStream(ssc, "192.168.15.131", 1234)val lineDStream = flumeEventDstream.map( e => {new String(e.event.getBody.array)})lineDStream.print()ssc.start()ssc.awaitTermination()}
}
测试:
1、启动Spark Streaming程序。
2、启动Flume。
bin/flume-ng agent -n a4 -f myagent/a4.conf -c conf -Dflume.root.logger=INFO,console
3、拷贝日志文件到/root/training/logs目录。
4、观察输出,采集到数据。
2. 基于Custom Sink的Pull模式
比第一种有更好的健壮性和容错性。生产使用这个方式。
不同于Flume直接将数据推送到Spark Streaming中,第二种模式通过以下条件运行一个正常的Flume Sink。Flume将数据推送到Sink中,并且数据保持buffered状态。Spark Streaming使用一个可靠的Flume接收器和转换器从Sink拉取数据。只要当数据被接收并且被Spark Streaming备份后,转换器才运行成功。这样,与第一种模式相比,保证了很好的健壮性和容错能力。然而,这种模式需要为Flume配置一个正常的Sink。
Flume 配置文件:
a1.channels = c1
a1.sinks = k1
a1.sources = r1a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/training/logsa1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 100000a1.sinks.k1.type = org.apache.spark.streaming.flume.sink.SparkSink
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = 192.168.15.131
a1.sinks.k1.port = 1234#组装source、channel、sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
Spark Streaming Full Demo:
注意除了需要使用Flume的lib的jar包以外,还需要
将Spark的jar包拷贝到Flume的lib目录下,spark-streaming-flume_2.10-2.1.0.jar也需要拷贝到Flume的lib目录下,同时加入IDEA工程的classpath。
import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.storage.StorageLevelobject FlumeLogPull {def main(args: Array[String]): Unit = {System.setProperty("hadoop.home.dir", "G:\\bin\\hadoop-2.5.2")Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)val conf = new SparkConf().setAppName("FlumeLogPull").setMaster("local[2]")val ssc = new StreamingContext(conf,Seconds(1))val flumeEvent = FlumeUtils.createPollingStream(ssc, "192.168.15.131", 1234, StorageLevel.MEMORY_ONLY_SER)val lineDStream = flumeEvent.map( e => {new String(e.event.getBody.array)})lineDStream.print()ssc.start()ssc.awaitTermination()}
}
测试:
1、启动Flume。
bin/flume-ng agent -n a1 -f myagent/a1.conf -c conf -Dflume.root.logger=INFO,console
2、启动Spark Streaming程序。
3、拷贝日志文件到/root/training/logs目录。
4、观察输出,采集到数据。
这篇关于Spark Streaming(三)—— 高级数据源Flume的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!