Spark的Streaming + Flume进行数据采集（flume主动推送或者Spark Stream主动拉取）

时间：2025-01-09 17:08:14

1、针对国外的开源技术，还是学会看国外的英文说明来的直接，迅速，这里简单贴一下如何看：

Spark的Streaming + Flume进行数据采集（flume主动推送或者Spark Stream主动拉取）

2、进入到flume的conf目录，创建一个flume-spark-push.sh的文件：

[hadoop@slaver1 conf]$ vim flume-spark-push.sh

Spark的Streaming + Flume进行数据采集（flume主动推送或者Spark Stream主动拉取）

配置一下这个文件，flume使用avro的。

# example.conf: A single-node Flume configuration

# Name the components on this agent

#定义这个agent中各组件的名字，给那三个组件sources，sinks，channels取个名字,是一个逻辑代号:

#a1是agent的代表。

a1.sources = r1

a1.channels = c1

a1.sinks = k1

# Describe/configure the source 描述和配置source组件：r1

#类型, 从网络端口接收数据,在本机启动, 所以localhost, type=spoolDir采集目录源,目录里有就采

#type是类型，是采集源的具体实现，这里是接受网络端口的，netcat可以从一个网络端口接受数据的。netcat在linux里的程序就是nc，可以学习一下。

#bind绑定本机localhost。port端口号为44444。

a1.sources.r1.type = exec

a1.sources.r1.bind = tail -f /home/hadoop/data_hadoop/spark-flume/wctotal.log

a1.sources.r1.shell = /bin/bash -c

# Describe the sink 描述和配置sink组件：k1

#type，下沉类型，使用logger，将数据打印到屏幕上面。

#a1.sinks.k1.type = logger

# Use a channel which buffers events in memory 描述和配置channel组件，此处使用是内存缓存的方式

#type类型是内存memory。

#下沉的时候是一批一批的, 下沉的时候是一个个eventChannel参数解释：

#capacity：默认该通道中最大的可以存储的event数量，1000是代表1000条数据。

#trasactionCapacity：每次最大可以从source中拿到或者送到sink中的event数量。

a1.channels.c1.type = memory

a1.channels.c1.capacity =

a1.channels.c1.transactionCapacity = 

# define sink

a1.sinks.k1.type= avro

a1.sinks.k1.hostname = slaver1

a1.sinks.k1.port = 

# Bind the source and sink to the channel 描述和配置source  channel   sink之间的连接关系

#将sources和sinks绑定到channel上面。

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

3、然后去Spark的github查看项目demo：https://github.com/apache/spark

具体案例如：https://github.com/apache/spark/blob/v1.5.1/examples/src/main/scala/org/apache/spark/examples/streaming/FlumeEventCount.scala

代码如下所示：

import org.apache.spark._

import org.apache.spark.streaming._

import org.apache.spark.streaming.StreamingContext._

import org.apache.spark.streaming.flume._

import org.apache.spark.util.IntParam

val ssc = new StreamingContext(sc, Seconds())

val stream = FlumeUtils.createStream(ssc, slaver1, , StorageLevel.MEMORY_ONLY_SER_2)

stream.count().map(cnt => "Received " + cnt + " flume events." ).print()

ssc.start()             // Start the computation

ssc.awaitTermination()  // Wait for the computation to terminate

导入flume的包的时候出现问题，找不到包：import org.apache.spark.streaming.flume._

scala> import org.apache.spark.streaming.flume._

<console>:: error: object flume is not a member of package org.apache.spark.streaming

       import org.apache.spark.streaming.flume._

由于没有搭建maven项目，在命令行需要导入jar包，这里先放置一下，稍后继续记笔记。

待续.......

相关文章