文章来自: http://lxw1234.com/?p=217
软件环境:
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
效果示例:
注意事项:
参见原文: http://lxw1234.com/?p=217
相关推荐
本科毕业设计项目,基于spark streaming+flume+kafka+hbase的实时日志处理分析系统 基于spark streaming+flume+kafka+hbase的实时日志处理分析系统 本科毕业设计项目,基于spark streaming+flume+kafka+hbase的...
基于spark streaming+flume+kafka+hbase的实时日志处理分析系统源码(分控制台版本和Web UI可视化版本).zip基于spark streaming+flume+kafka+hbase的实时日志处理分析系统源码(分控制台版本和Web UI可视化版本).zip...
基于spark streaming+flume+kafka+hbase的实时日志处理分析系统(分为控制台版本和基于s
【资源说明】 1、该资源包括项目的全部源码,下载可以直接使用! 2、本项目适合作为计算机、数学、电子...基于spark streaming+flume+kafka+hbase的实时日志处理分析系统源码(分为控制台版本和Web UI可视化版本).zip
1、资源内容:基于spark streaming+flume+kafka+hbase的实时日志处理分析系统(分为控制台版本和基于springboot、Echarts等的Web UI可视化版本) 2、代码特点:内含运行结果,不会运行可私信,参数化编程、参数可方便...
基于Spark Streaming + Kafka + Flume 实现的日志收集处理系统.zip
基于spark streaming+flume+kafka+hbase的实时日志处理分析系统源码(分控制台版本和Web UI可视化版本).zip 基于spark streaming+flume+kafka+hbase的实时日志处理分析系统源码(分控制台版本和Web UI可视化版本).zip ...
【资源说明】 1、该资源包括项目的全部源码,下载可以直接使用! 2、本项目适合作为计算机、数学、电子信息等专业的课程设计、期末大作业和毕设...基于spark streaming+kafka+hbase的日志统计分析系统源码+项目说明.zip
人工智能-spark
生成的数据主要是模拟某学习网站学习视频课程的访问量(其中*以“ / class”开头的表示实战课程,然后通过流水线Flume + Kafka + SparkStreaming进行实时日志的收集,HBase来存储数据)*注意事项(使用的软件工具及...
flume+Logstash+Kafka+Spark Streaming进行实时日志处理分析【大数据】
这个压缩包中包含的是Flume-1.6.0对接Spark-1.6.2下的Spark Streaming时需要用到的jar包,将这些jar包放置到flume安装目录下对应的lib中
spark-streaming-flume_2.11-2.1.0.jar
毕业设计,课程设计,项目源码均经过助教老师测试,运行无误,欢迎下载交流 ----- 下载后请首先打开README.md文件(如有)