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

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

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

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

配置一下这个文件,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 = 1000
a1.channels.c1.transactionCapacity = 100


# define sink
a1.sinks.k1.type= avro
a1.sinks.k1.hostname = slaver1
a1.sinks.k1.port = 9999


# 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(5))

val stream = FlumeUtils.createStream(ssc, slaver1, 9999, 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>:28: error: object flume is not a member of package org.apache.spark.streaming
       import org.apache.spark.streaming.flume._

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

待续.......

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏腾讯云数据库(TencentDB)

【腾讯云 MongoDB】 基于snapshot的从库读优化

我们发现腾讯云上一些腾讯云MongoDB实例在主库写压力比较大的情况下,这时从库上会出现很多慢查询,经过调查发现,从库在回放oplog的时候加了全局锁,阻塞了所...

6981
来自专栏喵了个咪的博客空间

[喵咪大数据]HUE大数据管理工具

日常的大数据使用中经常是在服务器命名行中进行操作,可视化功能仅仅依靠着各个组件自带的网页进行,那么有没有一个可以结合大家能在一个网页上的管理工具呢?答案是肯定的...

7598
来自专栏点滴积累

安装 Accumulo——突破自己,就是成长

前言 在我刚开始接触分布式集群的时候,是自己在几台虚拟机中手动安装的 Hadoop 和 Spark ,所以当时对 Hadoop 的配置有个简单的印象 ,但是后面...

50913
来自专栏技术小黑屋

一些关于加速Gradle构建的个人经验

目前绝大多数的Android项目都是基于Grale了,因为Gradle确实给我们带来了很多便利,然而,在使用了Gradle后,最大的不满就是编译起来太慢了。解决...

1651
来自专栏区块链

20种功能强大的跨平台渗透测试工具

什么是渗透测试?用来测试软件是否存在例如安全方面的漏洞,如果已经存在,会不会被入侵。 渗透测试流程: 1、列出软件或系统潜在的安全漏洞。 2、根据漏洞的严重性进...

2497
来自专栏Hadoop数据仓库

利用Flume将MySQL表数据准实时抽取到HDFS

一、为什么要用到Flume         在以前搭建HAWQ数据仓库实验环境时,我使用Sqoop抽取从MySQL数据库增量抽取数据到HDFS,然后用HAWQ的...

7158
来自专栏解Bug之路

解Bug之路-Druid的Bug 原

笔者很热衷于解决Bug,同时比较擅长(网络/协议)部分,所以经常被唤去解决一些网络IO方面的Bug。现在就挑一个案例出来,写出分析思路,以飨读者,希望读者在以后...

1535
来自专栏北京马哥教育

JMeter使用技巧

在这此对新版本jmeter的学习+温习的过程,发现了一些以前不知道的功能,所以,整理出来与大分享。本文内容如下。 如何使用英文界面的jmeter 如何使用镜像...

5229
来自专栏GreenLeaves

Oracle 客户端安装

Oracle 客户端的安装方式一种有两种: 1、Oracle标准客户端   点击下载 ? 这是Oracle提供的标准版11r2的客户端 2、Oracle Dat...

5378
来自专栏区块链入门

第4课 如何在UBUNTU虚拟机上编译EOS完成环境搭建?

【本文目标】 通过本文实践,能在WINDOWS操作系统搭建UBUNTU 18.04 LTS环境,并在此环境编译成功EOS v1.0.5正式发布版本。 【前置...

3315

扫码关注云+社区

领取腾讯云代金券