Spark Streaming+ FlumeNG整合

深藏阁楼爱情的钟 2022-08-04 09:21 145阅读 0赞

转自：/images/20220731/7061981288644f5eac35aaf731ffc7ee.png

SparkStreaming 提供一个处理Flume流的事件，

**\[html\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  class FlumeUtils  
2.    
3.  def createStream(ssc: StreamingContext, hostname: String, port: Int, storageLevel: StorageLevel = StorageLevel.MEMORY\_AND\_DISK\_SER\_2): ReceiverInputDStream\[SparkFlumeEvent\]  
4.    
5.  Create a input stream from a Flume source.

SparkStreaming的代码

**\[html\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  object FlumeEventPrint \{  
2.    def main(args: Array\[String\]) \{  
3.      /\*  
4.      if (args.length **<** **2**) \{  
5.        System.err.println(  
6.          "Usage: FlumeEventCount **<****host****>** **<****port****>**")  
7.        System.exit(1)  
8.      \}  
9.      \*/  
10.   
11.   
12.     //val Array(host, port) = args  
13.   
14.     val batchInterval = Milliseconds(2000)  
15.   
16.     // Create the context and set the batch size  
17.     val sparkConf = new SparkConf().setAppName("FlumeEventCount").setMaster("local")  
18.     val ssc = new StreamingContext(sparkConf, batchInterval)  
19.   
20.     // Create a flume stream  
21.     val stream = FlumeUtils.createStream(ssc, "localhost", 33333, StorageLevel.MEMORY\_AND\_DISK)  
22.     //stream.saveAsTextFiles("nothing")  
23.     // Print out the count of events received from this server in each batch  
24.     //stream.map\{ cnt =**>** "Received " + cnt mkString " " + " flume events." \}.saveAsTextFiles("nothing/")  
25.     stream.map(e=**>**e.event)saveAsTextFiles("nothing/")  
26.   
27.     ssc.start()  
28.     ssc.awaitTermination(10000)  
29.     ssc.stop(true,true)  
30.   \}  
31. \}

这段代码是将从Flume Sink发送的事件打印到文本中，flume sink会根据配置的host与port将事件发送到spark

然后需要一个Flume的Agent来将数据接收，暂存及转发，下面是flume Agent的配置文件

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

对于我来说，Flume的框架不是很容易理解，而且容易产生误解.(之前我认为sink是用来处理数据的，后来发现sink也只是用来转发数据事件的，最终处理事件的是另外的外部程序，严格来说，Flume只是将数据封装与传送，不会对内容有任何的业务处理)

![36634059][]

通过我的配置文件。

Source:配置成一个avro的服务器(接收来自avro客户端的事件), host为local host, port 为44444

Channel:通过内存来暂存事件

Sink:配置成一个avro的客户端(向host为localhost,端口为33333的avro服务器发送事件), 这里avro的服务器即是spark streaming的程序.

启动的顺序，先启动spark streaming的程序，然后程序Flume agent, 最后再额外启动一个avro客户端向Flume agent的Source发送数据。

命令：

启动flume agent: bin/flume-ng agent --conf conf --conf-file conf/agent1.conf --name a1 -Dflume.root.logger=INFO,console

启动avro client: bin/flume-ng avro-client --conf conf -H localhost -p 44444 -F /Users/heruilong/Documents/subway/gxy\_9.xml -Dflume.root.logger=DEBUG,console  (flume内部提供了一个avro client的实现)

数据结果:(我传送的是一个xml文件)

打印出来的是json格式

\{"headers": \{\}, "body": \{"bytes": "<?xml version="1.0" encoding="GBK"?>"\}\}  
\{"headers": \{\}, "body": \{"bytes": ""\}\}  
\{"headers": \{\}, "body": \{"bytes": "<subwaycard>"\}\}  
\{"headers": \{\}, "body": \{"bytes": "  <card ID="10007510033252741">"\}\}  
\{"headers": \{\}, "body": \{"bytes": "    <recha>"\}\}  
\{"headers": \{\}, "body": \{"bytes": "      <recharge/>"\}\}  
\{"headers": \{\}, "body": \{"bytes": "    </recha>"\}\}  
\{"headers": \{\}, "body": \{"bytes": "    <cons>"\}\}

需要参考的spark接口

**\[html\]**  [view plain][] [copy][view plain] [![在CODE上查看代码片][CODE]][CODE_CODE] [![派生到我的代码片][ico_fork.svg]][ico_fork.svg 1]

1.  class SparkFlumeEvent extends Externalizable   
2.

[view plain]: /images/20220731/7061981288644f5eac35aaf731ffc7ee.png#
[CODE]: https://code.csdn.net/assets/CODE_ico.png
[CODE_CODE]: https://code.csdn.net/snippets/415101
[ico_fork.svg]: https://code.csdn.net/assets/ico_fork.svg
[ico_fork.svg 1]: https://code.csdn.net/snippets/415101/fork
[36634059]: /images/20220731/7061981288644f5eac35aaf731ffc7ee.png