文章来自: 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
相关推荐
3. spark-streaming-flume-sink_2.10-1.6.1.jar:这是 Spark Streaming 与 Flume 对接的关键组件,称为 Flume 收集器(sink)。这个 jar 包实现了将 Flume 接收到的数据流发送到 Spark Streaming 进行处理的接口。...
- **Spark SQL**:用于离线分析,可以方便地执行SQL查询,对接多种数据源,包括HDFS、Hive、Cassandra等,帮助我们从海量日志中提取关键信息,例如用户购买行为、热门商品等。 - **Spark Streaming**:处理实时...
实时数据处理:Kafka + Spark Streaming 数据应用层:MLlib 产生一个模型 als算法 数据展示和对接:Zeppelin 选用考量: HDFS不管是在存储的性能,稳定性 吞吐量 都是在主流文件系统中很占有优势的 如果感觉HDFS...
标题中的“Flume push数据到SparkStreaming”是指在大数据处理领域中,使用Apache Flume将实时数据流推送到Apache Spark Streaming进行进一步的实时分析和处理的过程。这两个组件都是Apache Hadoop生态系统的重要...
sparkstreaming把统计结果写入到数据库里面 hbase表设计:create 'aqy_video_clickcount','info' RowKey设计:day_videoid 功能二: 统计今天到现在为止从搜索引擎引流过来的实战课程的访问量 功能1+从搜索引擎...
因此,我们需要自己编写定制的接收器或者利用现有的第三方库来实现MetaQ与Spark Streaming的对接。 一种常见的方法是通过Apache Kafka作为桥梁,因为MetaQ和Spark都支持与Kafka的集成。你可以将MetaQ中的消息发布到...
- **Spark Streaming**:用于实时数据流处理,它以微批处理的方式处理数据,支持多种数据源如Kafka、Flume等。 2. **Echarts**: - **Echarts简介**:Echarts是由百度开发的一款基于JavaScript的开源可视化图表库...
12-8 -通过定时调度工具每一分钟产生...2.对接python日志产生器输出的日志到Flume 定义名字为streaming_project.conf 选型:access.log ==>控制台输出 exec memory logger streaming_project.conf文件具体配置:
5. **Spark SQL**:Spark SQL提供了一种统一的方式来查询结构化和半结构化数据,可以与Hive、JDBC/ODBC接口等外部数据源集成,使得Spark能与传统的SQL环境无缝对接。 6. **Spark Streaming**:Spark Streaming是...
在 Spark 2.4.0 中,它增强了对接 Kafka、Flume、Twitter 等数据源的稳定性和性能。DStream(Discretized Stream)是 Spark Streaming 提供的基本抽象,允许用户以类似批处理的方式处理流数据。 GraphX 是 Spark ...
它可以与Kafka、Flume、Twitter等数据源对接,广泛应用于日志分析、监控系统等场景。 5. **MLlib**: MLlib是Spark的机器学习库,提供了多种算法,如分类、回归、聚类、协同过滤等,并且支持管道和模型选择,方便...
2. 扩展性:Spark可以无缝对接Hadoop生态系统,支持多源数据接入和处理。 3. 易用性:Spark提供了丰富的API和SQL支持,简化了开发工作。 五、应用场景 本系统可广泛应用于电商、社交媒体、广告推荐等领域,帮助...
它可以对接多种数据源,如Kafka、Flume和Twitter等。 5. **MLlib**:Spark的机器学习库提供了大量常用的机器学习算法,包括分类、回归、聚类、协同过滤等,同时提供了模型选择、调优和评估工具。 6. **GraphX**:...
实时数据处理:Kafka + Spark Streaming 数据应用层:MLlib 产生一个模型 als算法 数据展示和对接:Zeppelin 选用考量: HDFS不管是在存储的性能,稳定性 吞吐量 都是在主流文件系统中很占有优势的 如果感觉HDFS...
它能够与 Kafka、Flume 等数据源无缝对接,实时处理海量数据流。 6. Machine Learning 库 MLlib Spark 3.1.1 的 MLlib 提供了丰富的机器学习算法,如分类、回归、聚类、协同过滤等,支持模型训练和评估。结合 ...
Spark的核心组件包括:Spark Core(基础框架)、Spark SQL(SQL和数据处理)、Spark Streaming(流处理)、MLlib(机器学习库)和GraphX(图计算)。在本项目中,我们将主要利用Spark Core和Spark SQL进行数据处理。...
* 整体架构:日志文件、Flume、Kafka、Spark Streaming、启动日志Topic、页面访问Topic、动作日志Topic、曝光日志Topic、错误日志Topic、DWD层、ODS层、Kafka * 采集日志数据:上传模拟日志数据生成器到/opt/module/...
实时推荐服务使用 Spark Streaming 作为实时推荐系统,通过接收 Kafka 中缓存的数据,通过设计的推荐算法实现对实时推荐的数据处理,并将结构合并更新到 MongoDB 数据库中。 数据模型 该系统的数据模型主要包括 ...
- **Spark Streaming**:用于实时流数据处理,可以处理来自多种源的数据流,如Kafka、Flume等。 - **MLlib**:机器学习库,包括多种机器学习算法,如分类、回归、聚类、协同过滤等。 - **GraphX**:用于图计算,...
Spark Core提供了分布式任务调度和内存计算的基础,而Spark SQL则负责处理结构化数据,可与SQL语法无缝对接。Spark Streaming处理实时流数据,MLlib包含大量机器学习算法,GraphX则用于图数据的处理和分析。 二、...