这里写一个flume整合sparkstreaming的例子
我这里使用scala ide和maven的方式
spark用的1.2版本,scala是2.10版本,flume使用最新版的1.6
整合的第一步,要加上flume的maven依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-flume_2.10</artifactId>
<version>1.2.1</version>
</dependency>
在这里,不进行spark环境的搭建,为了方便起见,直接使用scala ide本地local模式运行
代码就已单词计数为例
package cn.han
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.flume._
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.storage.StorageLevel
object MySpark {
def main(args: Array[String]): Unit = {
val sc = new SparkContext("local[2]", "Spark Streaming Flume Integration")
//创建StreamingContext,3秒一个批次
val ssc = new StreamingContext(sc, Seconds(3))
val hostname = "192.168.5.102"
val port = 11111
val flumeStream = FlumeUtils.createStream(ssc, hostname, port)
val rss2=flumeStream.flatMap(x =>{
val by=x.event.getBody.array()
val str=new String(by)
val sb=str.split(" ")
sb
})
val rdd3=rss2.map(x=>(x,1))
val rdd4=rdd3.reduceByKey(_+_)
rdd4.print()
//开始运行
ssc.start()
//计算完毕退出
ssc.awaitTermination()
sc.stop()
}
}
这里有一个需要及其注意的地方,就是local的参数,必须要大于1,除了receive也需要占用一个,如果写一个,就接收不到数据
然后,是flume的一些配置,首先,下载最新版的flume,解压,直接在conf目录下新建
agent3
直接启动flume
bin/flume-ng agent --conf ./conf/ -f conf/agent3 -Dflume.root.logger=DEBUG,console -n agent3
flume这边就启动完成了
然后还需要一个flume的源就是source,这里配置的是avro,就使用log4j的方式直接写入flume
首先引入log4j放入maven,这里面需要下载一个jar包引入才行
flume-ng-log4jappender-1.5.0-cdh5.1.3-jar-with-dependencies.jar
然后需要修改下log4j.properties文件
log4j.rootLogger=INFO,flume
log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname = 192.168.5.220
log4j.appender.flume.Port = 41414
log4j.appender.flume.UnsafeMode = true
这时候在写个简单的测试类进行测试
package cn.han;
import org.apache.log4j.Logger;
public class MyLogh {
private static Logger logger = Logger.getLogger(MyLogh.class);
public static void main(String[] args) throws Exception{
while(true){
for(int i=0;i<120;i++){
logger.info("hello you");
}
Thread.sleep(1000);
}
}
}
这样运行spark个程序就可以看到结果,每3秒执行一次job
- 大小: 8.8 KB
- 大小: 98.8 KB
分享到:
相关推荐
1.Spark Streaming整合Flume需要的安装包. 2. Spark Streaming拉取Flume数据的flume配置文件.conf 3. Flume向Spark Streaming推数据的flume配置文件.conf
通过flume监控文件,让kafka消费flume数据,再将sparkstreaming连接kafka作为消费者进行数据处理,文档整理实现
第1章 课程介绍 第2章 初识实时流处理 第3章 分布式日志收集框架Flume 第4章 分布式发布...第9章 Spark Streaming整合Flume 第10章 Spark Streaming整合Kafka 第11章 Spark Streaming整合Flume&Kafka打造通用流处理基础
Spark_Streaming整合Flume.md
第1章 课程介绍课程介绍 第2章 初识实时流处理 ...第9章 Spark Streaming整合Flume 第10章 Spark Streaming整合Kafka 第11章 Spark Streaming整合Flume&Kafka;打造通用流处理基础 第12章 Spark Streaming项目实战
1.Flume Agent的编写: flume_push_streaming.conf 2.代码 4.本地测试总结 1.Flume Agent的编写: flum
Spark Straming 提供了以下两种方式用于 Flume 的整合。 在推送式方法 (Flume-style Push-based Approach) 中,Spark Streaming 程序需要对某台服务器的某个端口进行监听,Flume 通过 `avro Sink` 将数据源源不断推...
将该jar包上传至flume/lib目录下,并将spark-streaming-flume其他版本jar包删除即可使用,该jar包适用于spark2.1.3版本使用
这是一个大数据实时流处理分析系统 Demo,实现对用户日志的实时分析,采用 Flume + kafka + SparkStreaming + Hbase + SSM + Echarts 的架构。 主要内容包括: 编写 python 脚本,模拟源源不断产生网站的用户行为...
第2部分(第8~11章)讲解Spark的原生态组件,包括SparkCore、SparkSQL、SparkStreaming、DataFrame,以及介绍Scala、SparkAPI、SparkSQL、SparkStreaming、DataFrame原理和CDH版本环境下实战操作,其中Flume和Kafka...
这是一个大数据实时流处理分析系统 Demo,实现对用户日志的实时分析,采用 Flume + kafka + SparkStreaming + Hbase + SSM + Echarts 的架构。 主要内容包括: 编写 python 脚本,模拟源源不断产生网站的用户行为...
Storm系统与Spark Streaming系统共同部署在YARN上,完成了平台的整合与数据资源共享,节省了网络开销。解决精准性营销的问题:通过对电信的DPI数据进行分析,建立用户标签库,对用户进行画像。通过数据挖掘算法发现目标...
同时 Spark 还可以让使用者们简单而低消耗地将各种处理流程 整合到一起,这样既满足了对实时计算以及实时处 理的要求, 还减轻了对各种平台分别管理的负担[1]。 1 Spark 的五大组件 在 spark Core 中实现了 Spark 的...
Streaming,入侵检测采用Spark MLlib,数据存储使用HDFS与Redis,可视化采用Flask、SocketIO、Echarts、Bootstrap 项目使用 简易使用说明 开箱即用,直接在v2目录中使用 python3 app.py # 开启后直接访问,即可...
接着通过Structured Streaming整合Kafka,读取Kafka消息队列的实时流数据,并与mysql进行集成,将读取到的数据储存到MySQL数据库当中,通过编写java程序读取MySQL中的数据和编写Web数据处理程序将数据传至前端,最后...
整合Flume和Kafka完成实时数据采集 2 Kafka概述 官网:http://kafka.apache.org/ A distribute streaming platform. PUBLISH & SUBSCRIBE(订阅和发布): Read and write streams of data like a messaging system...