微信搜索lxw1234bigdata | 邀请体验:数阅–数据管理、OLAP分析与可视化平台 | 赞助作者:赞助作者

Spark Streaming+Flume对接实验

Spark lxw1234@qq.com 13390℃ 4评论

软件环境:

flume-ng-core-1.4.0-cdh5.0.0

spark-1.2.0-bin-hadoop2.3

流程说明:

  • Spark Streaming: 使用spark-streaming-flume_2.10-1.2.0插件,启动一个avro source,用来接收数据,并做相应的处理;
  • Flume agent:source监控本地文件系统的一个目录,当文件发生变化时候,由avro sink发送至Spark Streaming的监听端口

Flume配置:

flume-lxw-conf.properties

	
#-->设置sources名称
agent_lxw.sources = sources1
#--> 设置channel名称
agent_lxw.channels = fileChannel
#--> 设置sink 名称 
agent_lxw.sinks = sink1

# source 配置
## 一个自定义的Source,实现类似tail -f 的功能,比exec source更可靠
agent_lxw.sources.sources1.type = org.apache.flume.source.taildirectory.DirectoryTailSource
agent_lxw.sources.sources1.dirs = lxwlog
## 监控的目录
agent_lxw.sources.sources1.dirs.lxwlog.path = file:///tmp/lxw-source
#监控文件的正则规则,此正则用java的正则
agent_lxw.sources.sources1.dirs.lxwlog.file-pattern = ^lxw_.*log$
agent_lxw.sources.sources1.first-line-pattern = ^(.*)$
agent_lxw.sources.sources1.channels = fileChannel


# sink 1 配置 将数据发送至slave004.lxw1234.com的44444端口
agent_lxw.sinks.sink1.type = avro
agent_lxw.sinks.sink1.hostname = slave004.lxw1234.com
agent_lxw.sinks.sink1.port = 44444
agent_lxw.sinks.sink1.channel = fileChannel
agent_lxw.sinks.sink1.batch-size = 500
agent_lxw.sinks.sink1.connect-timeout = 40000
agent_lxw.sinks.sink1.request-timeout = 40000

agent_lxw.channels.fileChannel.type = file
#-->检测点文件所存储的目录
agent_lxw.channels.fileChannel.checkpointDir = /tmp/flume/checkpoint/site
#-->数据存储所在的目录设置
agent_lxw.channels.fileChannel.dataDirs = /tmp/flume/data/site
#-->隧道的最大容量
agent_lxw.channels.fileChannel.capacity = 10000
#-->事务容量的最大值设置
agent_lxw.channels.fileChannel.transactionCapacity = 100

Spark Streaming程序:

Spark_Flume.scala

 

	
package com.lxw.test

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.flume.FlumeUtils


object Spark_Flume {
  def main (args : Array[String]) {
    
    if(args.length < 2) {
      println("Usage: Spark_Flume <hostname> <port>")
      System.exit(1)
    }
    
    val hostname = args(0)
    val port = Integer.parseInt(args(1))
    
    val sc = new SparkContext(new SparkConf().setAppName("Spark_Flume"))
    val ssc = new StreamingContext(sc, Seconds(10))
    val flumeStream = FlumeUtils.createStream(ssc, hostname, port,StorageLevel.MEMORY_AND_DISK)
    
    flumeStream.map(e => "Event:header:" + e.event.get(0).toString + "body: " + new String(e.event.getBody.array)).print()
    ssc.start()
    ssc.awaitTermination()
    
  }
}

启动:

  • 先启动Spark Streaming程序:
	
./spark-submit \
	--name "spark-flume" \
	--master spark://192.168.1.130:7077 \
	--executor-memory 1G \
	--class com.lxw.test.Spark_Flume \
	/home/liuxiaowen/spark-flume.jar slave004.lxw1234.com 44444
  • 再启动Flume agent:
	
flume-ng agent -n agent_lxw --conf . -f flume-lxw-conf.properties

效果示例:

命令行往文件中增加数据

Spark and Flume

Spark and Flume

 

Flume监听到文件变化

Spark and Flume

Spark and Flume

 

Spark Streaming接收并处理数据

Spark and Flume

Spark and Flume

 

注意事项:

  1. Spark集群已经部署好,采用Standalone模式;
  2. Spark集群中每台节点需要将spark-streaming-flume_2.10-1.2.0.jar和flume-avro-source-1.4.0-cdh5.0.0.jar添加至SPARK_CLASSPATH中;
  3. Spark_Flume.scala在编译时候依赖:spark-assembly-1.2.0-hadoop2.3.0.jar、spark-streaming-flume_2.10-1.2.0.jar、flume-avro-source-1.4.0-cdh5.0.0.jar、flume-ng-sdk-1.4.0-cdh5.0.0.jar;
  4. 启动Spark Streaming时候传入的hostname (slave004.lxw1234.com),必须是Spark集群中的一台节点,Spark会在这台机器上启动NettyServer;

 

如果觉得本博客对您有帮助,请 赞助作者

转载请注明:lxw的大数据田地 » Spark Streaming+Flume对接实验

喜欢 (12)
分享 (0)
发表我的评论
取消评论
表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址
(4)个小伙伴在吐槽
  1. lz,spark streaming flume 整合 我遇到了问题,spark无法绑定端口. 错误信息如下: 15/11/21 03:43:36 WARN TaskSetManager: Lost task 0.0 in stage 9.0 (TID 84, 192.168.111.131): org.jboss.netty.channel.ChannelException: Failed to bind to: /192.168.111.130:33233 at org.jboss.netty.bootstrap.ServerBootstrap.bind(ServerBootstrap.java:272) at org.apache.avro.ipc.NettyServer.(NettyServer.java:106) at org.apache.avro.ipc.NettyServer.(NettyServer.java:119) at org.apache.avro.ipc.NettyServer.(NettyServer.java:74) at org.apache.avro.ipc.NettyServer.(NettyServer.java:68) at org.apache.spark.streaming.flume.FlumeReceiver.server$lzycompute(FlumeInputDStream.scala:142) at org.apache.spark.streaming.flume.FlumeReceiver.server(FlumeInputDStream.scala:142) at org.apache.spark.streaming.flume.FlumeReceiver.onStart(FlumeInputDStream.scala:145) at org.apache.spark.streaming.receiver.ReceiverSupervisor.startReceiver(ReceiverSupervisor.scala:148) at org.apache.spark.streaming.receiver.ReceiverSupervisor.start(ReceiverSupervisor.scala:130) at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:542) at org.apache.spark.streaming.scheduler.ReceiverTracker$ReceiverTrackerEndpoint$$anonfun$9.apply(ReceiverTracker.scala:532) at org.apache.spark.SparkContext$$anonfun$37.apply(SparkContext.scala:1986) at org.apache.spark.SparkContext$$anonfun$37.apply(SparkContext.scala:1986) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) at org.apache.spark.scheduler.Task.run(Task.scala:88) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: java.net.BindException: Cannot assign requested address at sun.nio.ch.Net.bind0(Native Method) at sun.nio.ch.Net.bind(Net.java:433) at sun.nio.ch.Net.bind(Net.java:425) at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223) at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74) at org.jboss.netty.channel.socket.nio.NioServerBoss$RegisterTask.run(NioServerBoss.java:193) at org.jboss.netty.channel.socket.nio.AbstractNioSelector.processTaskQueue(AbstractNioSelector.java:372) at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:296) at org.jboss.netty.channel.socket.nio.NioServerBoss.run(NioServerBoss.java:42) ... 3 more 看上去好像是slave节点无法绑定master节点端口 改成localhost绑定,这个错误就不出现。 但是这样的话slave节点的flumeStream应该就没有数据。 还有个问题是FlumeUtils.createStream 在执行的时候是绑定端口的,执行完一个批次过后应该就把端口释放了的,这个我是看flume日志猜测的,不知道是不是这样。而且spark程序整个执行过程中通过netstat | grep 看不到正在执行的端口. flume日志: 2015-11-21 03:52:27,936 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.AbstractRpcSink.createConnection(AbstractRpcSink.java:206)] Rpc sink sinks1: Building RpcClient with hostname: master, port: 33233 2015-11-21 03:52:27,936 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.AvroSink.initializeRpcClient(AvroSink.java:126)] Attempting to create Avro Rpc client. 2015-11-21 03:52:27,937 (SinkRunner-PollingRunner-DefaultSinkProcessor) [WARN - org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:634)] Using default maxIOWorkers 2015-11-21 03:52:29,292 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:160)] Unable to deliver event. Exception follows. org.apache.flume.EventDeliveryException: Failed to send events at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:392) at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68) at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147) at java.lang.Thread.run(Thread.java:745) Caused by: org.apache.flume.FlumeException: NettyAvroRpcClient { host: master, port: 33233 }: RPC connection error at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:182) at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:121) at org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:638) at org.apache.flume.api.RpcClientFactory.getInstance(RpcClientFactory.java:89) at org.apache.flume.sink.AvroSink.initializeRpcClient(AvroSink.java:127) at org.apache.flume.sink.AbstractRpcSink.createConnection(AbstractRpcSink.java:211) at org.apache.flume.sink.AbstractRpcSink.verifyConnection(AbstractRpcSink.java:272) at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:349) ... 3 more Caused by: java.io.IOException: Error connecting to master/192.168.111.130:33233 at org.apache.avro.ipc.NettyTransceiver.getChannel(NettyTransceiver.java:261) at org.apache.avro.ipc.NettyTransceiver.(NettyTransceiver.java:203) at org.apache.avro.ipc.NettyTransceiver.(NettyTransceiver.java:152) at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:168) ... 10 more Caused by: java.net.ConnectException: Connection refused at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.connect(NioClientSocketPipelineSink.java:496) at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.processSelectedKeys(NioClientSocketPipelineSink.java:452) at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.run(NioClientSocketPipelineSink.java:365) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ... 1 more 2015-11-21 03:52:34,344 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.AbstractRpcSink.createConnection(AbstractRpcSink.java:206)] Rpc sink sinks1: Building RpcClient with hostname: master, port: 33233 2015-11-21 03:52:34,344 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.AvroSink.initializeRpcClient(AvroSink.java:126)] Attempting to create Avro Rpc client. 2015-11-21 03:52:34,345 (SinkRunner-PollingRunner-DefaultSinkProcessor) [WARN - org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:634)] Using default maxIOWorkers 2015-11-21 03:52:34,483 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:160)] Unable to deliver event. Exception follows. org.apache.flume.EventDeliveryException: Failed to send events at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:392) at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68) at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147) at java.lang.Thread.run(Thread.java:745) Caused by: org.apache.flume.FlumeException: NettyAvroRpcClient { host: master, port: 33233 }: RPC connection error at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:182) at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:121) at org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:638) at org.apache.flume.api.RpcClientFactory.getInstance(RpcClientFactory.java:89) at org.apache.flume.sink.AvroSink.initializeRpcClient(AvroSink.java:127) at org.apache.flume.sink.AbstractRpcSink.createConnection(AbstractRpcSink.java:211) at org.apache.flume.sink.AbstractRpcSink.verifyConnection(AbstractRpcSink.java:272) at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:349) ... 3 more Caused by: java.io.IOException: Error connecting to master/192.168.111.130:33233 at org.apache.avro.ipc.NettyTransceiver.getChannel(NettyTransceiver.java:261) at org.apache.avro.ipc.NettyTransceiver.(NettyTransceiver.java:203) at org.apache.avro.ipc.NettyTransceiver.(NettyTransceiver.java:152) at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:168) ... 10 more Caused by: java.net.ConnectException: Connection refused at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.connect(NioClientSocketPipelineSink.java:496) at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.processSelectedKeys(NioClientSocketPipelineSink.java:452) at org.jboss.netty.channel.socket.nio.NioClientSocketPipelineSink$Boss.run(NioClientSocketPipelineSink.java:365) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    iDL2015-11-21 19:59 回复
    • 这种问题一般是hostname和ip的映射造成的,/etc/hosts文件配置。 一般在配置时候使用hostname,然后在/etc/hosts文件中配置和ip的映射。
      lxw1234@qq.com2015-11-23 14:16 回复
    • Hello,我也遇到了同样的问题,不过感觉应该不是hosts跟IP的映射导致的。请问你后来解决了吗?
      ZP2016-07-20 11:57 回复
  2. 我现在是spark程序启动成功,假设我这边spark的端口号是:9999...但是flume监听这个端口的时候,会出现 org.apache.flume.FlumeException: NettyAvroRpcClient { host: 119.129.70.177, port: 44444 }: RPC connection error 目前暂时不知道是不是我spark虽然没有报错,但是实际上是没有启动成功的....
    掂吾掂2016-11-01 20:06 回复