`
superlxw1234
  • 浏览: 550752 次
  • 性别: Icon_minigender_1
  • 来自: 西安
博客专栏
Bd1c0a0c-379a-31a8-a3b1-e6401e2f1523
Hive入门
浏览量:44355
社区版块
存档分类
最新评论

Spark Streaming+Flume对接实验

阅读更多

文章来自: 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

 

0
0
分享到:
评论

相关推荐

    Flume对接Spark Streaming的相关jar包

    3. spark-streaming-flume-sink_2.10-1.6.1.jar:这是 Spark Streaming 与 Flume 对接的关键组件,称为 Flume 收集器(sink)。这个 jar 包实现了将 Flume 接收到的数据流发送到 Spark Streaming 进行处理的接口。...

    基于Flume+kafka+spark大型电商网站日志分析系统(离线+实时).zip

    - **Spark SQL**:用于离线分析,可以方便地执行SQL查询,对接多种数据源,包括HDFS、Hive、Cassandra等,帮助我们从海量日志中提取关键信息,例如用户购买行为、热门商品等。 - **Spark Streaming**:处理实时...

    基于spark+flume+kafka+hbase的实时日志处理分析系统.zip

    实时数据处理:Kafka + Spark Streaming 数据应用层:MLlib 产生一个模型 als算法 数据展示和对接:Zeppelin 选用考量: HDFS不管是在存储的性能,稳定性 吞吐量 都是在主流文件系统中很占有优势的 如果感觉HDFS...

    Flume push数据到SparkStreaming

    标题中的“Flume push数据到SparkStreaming”是指在大数据处理领域中,使用Apache Flume将实时数据流推送到Apache Spark Streaming进行进一步的实时分析和处理的过程。这两个组件都是Apache Hadoop生态系统的重要...

    spark streaming实时网站分析项目实战.rar

    sparkstreaming把统计结果写入到数据库里面 hbase表设计:create 'aqy_video_clickcount','info' RowKey设计:day_videoid 功能二: 统计今天到现在为止从搜索引擎引流过来的实战课程的访问量 功能1+从搜索引擎...

    metaQ向spark传数据

    因此,我们需要自己编写定制的接收器或者利用现有的第三方库来实现MetaQ与Spark Streaming的对接。 一种常见的方法是通过Apache Kafka作为桥梁,因为MetaQ和Spark都支持与Kafka的集成。你可以将MetaQ中的消息发布到...

    基于spark+echarts实现的互联网行业数据大屏分析源码+项目说明.zip

    - **Spark Streaming**:用于实时数据流处理,它以微批处理的方式处理数据,支持多种数据源如Kafka、Flume等。 2. **Echarts**: - **Echarts简介**:Echarts是由百度开发的一款基于JavaScript的开源可视化图表库...

    SparkStreaming学习札记4-2020-2-15–SparkStreaming实时流处理项目实战

    12-8 -通过定时调度工具每一分钟产生...2.对接python日志产生器输出的日志到Flume 定义名字为streaming_project.conf 选型:access.log ==&gt;控制台输出  exec  memory  logger streaming_project.conf文件具体配置:

    Spark编程指南中文版

    5. **Spark SQL**:Spark SQL提供了一种统一的方式来查询结构化和半结构化数据,可以与Hive、JDBC/ODBC接口等外部数据源集成,使得Spark能与传统的SQL环境无缝对接。 6. **Spark Streaming**:Spark Streaming是...

    spark-2.4.0-bin-without-hadoop.tgz

    在 Spark 2.4.0 中,它增强了对接 Kafka、Flume、Twitter 等数据源的稳定性和性能。DStream(Discretized Stream)是 Spark Streaming 提供的基本抽象,允许用户以类似批处理的方式处理流数据。 GraphX 是 Spark ...

    Spark案例源码和数据集.rar

    它可以与Kafka、Flume、Twitter等数据源对接,广泛应用于日志分析、监控系统等场景。 5. **MLlib**: MLlib是Spark的机器学习库,提供了多种算法,如分类、回归、聚类、协同过滤等,并且支持管道和模型选择,方便...

    计算机课程毕设:基于Spark的行为日志分析系统设计与实现.zip

    2. 扩展性:Spark可以无缝对接Hadoop生态系统,支持多源数据接入和处理。 3. 易用性:Spark提供了丰富的API和SQL支持,简化了开发工作。 五、应用场景 本系统可广泛应用于电商、社交媒体、广告推荐等领域,帮助...

    Spark快速数据处理

    它可以对接多种数据源,如Kafka、Flume和Twitter等。 5. **MLlib**:Spark的机器学习库提供了大量常用的机器学习算法,包括分类、回归、聚类、协同过滤等,同时提供了模型选择、调优和评估工具。 6. **GraphX**:...

    基于Flume&spark&Flask的分布式实时日志分析与入侵检测系统.zip

    实时数据处理:Kafka + Spark Streaming 数据应用层:MLlib 产生一个模型 als算法 数据展示和对接:Zeppelin 选用考量: HDFS不管是在存储的性能,稳定性 吞吐量 都是在主流文件系统中很占有优势的 如果感觉HDFS...

    spark-3.1.1-bin-hadoop2.7

    它能够与 Kafka、Flume 等数据源无缝对接,实时处理海量数据流。 6. Machine Learning 库 MLlib Spark 3.1.1 的 MLlib 提供了丰富的机器学习算法,如分类、回归、聚类、协同过滤等,支持模型训练和评估。结合 ...

    基于Spark的行为日志分析系统设计与实现.zip

    Spark的核心组件包括:Spark Core(基础框架)、Spark SQL(SQL和数据处理)、Spark Streaming(流处理)、MLlib(机器学习库)和GraphX(图计算)。在本项目中,我们将主要利用Spark Core和Spark SQL进行数据处理。...

    01大数据项目之Spark实时(数据采集)

    * 整体架构:日志文件、Flume、Kafka、Spark Streaming、启动日志Topic、页面访问Topic、动作日志Topic、曝光日志Topic、错误日志Topic、DWD层、ODS层、Kafka * 采集日志数据:上传模拟日志数据生成器到/opt/module/...

    尚硅谷大数据技术之电商推荐系统.doc

    实时推荐服务使用 Spark Streaming 作为实时推荐系统,通过接收 Kafka 中缓存的数据,通过设计的推荐算法实现对实时推荐的数据处理,并将结构合并更新到 MongoDB 数据库中。 数据模型 该系统的数据模型主要包括 ...

    spark-2.1.1-bin-hadoop2.6.tgz

    - **Spark Streaming**:用于实时流数据处理,可以处理来自多种源的数据流,如Kafka、Flume等。 - **MLlib**:机器学习库,包括多种机器学习算法,如分类、回归、聚类、协同过滤等。 - **GraphX**:用于图计算,...

    aasPractice:《spark高级数据分析》练习

    Spark Core提供了分布式任务调度和内存计算的基础,而Spark SQL则负责处理结构化数据,可与SQL语法无缝对接。Spark Streaming处理实时流数据,MLlib包含大量机器学习算法,GraphX则用于图数据的处理和分析。 二、...

Global site tag (gtag.js) - Google Analytics