Spark Streaming(三)—— 高级数据源Flume

2024-06-19 04:38

本文主要是介绍Spark Streaming(三)—— 高级数据源Flume,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • 高级数据源Flume
    • 1. Push方式
    • 2. 基于Custom Sink的Pull模式

高级数据源Flume

Spark Streaming 是一个流式计算引擎,就需要对接外部数据源来对接、接收数据。每一个输入流DStream和一个Receiver对象相关联,这个Receiver从源中获取数据,并将数据存入内存中用于处理。Spark Streaming的基本数据源(文件流、RDD队列流、套接字流)上篇已经介绍过了,而Spark Streaming的高级数据流主要有Kafka,Flume,Kinesis,Twitter等。本文主要介绍Flume作为高级数据源的使用。

1. Push方式

Flume将数据推送给Spark Streaming。在这种方式下,Spark Streaming可以很方便的建立一个Receiver,起到一个Avro agent的作用。Flume可以将数据推送到该Receiver。

Flume配置文件:

#定义agent名, source、channel、sink的名称
a4.sources = r1
a4.channels = c1
a4.sinks = k1#具体定义source 采集该目录下的日志
a4.sources.r1.type = spooldir
a4.sources.r1.spoolDir = /root/training/logs#具体定义channel
a4.channels.c1.type = memory
a4.channels.c1.capacity = 10000
a4.channels.c1.transactionCapacity = 100#具体定义sink
a4.sinks = k1
a4.sinks.k1.type = avro
a4.sinks.k1.channel = c1
a4.sinks.k1.hostname = 192.168.15.131
a4.sinks.k1.port = 1234#组装source、channel、sink
a4.sources.r1.channels = c1
a4.sinks.k1.channel = c1

Spark Streaming Demo:
注意除了需要使用Flume的lib的jar包以外,还需要spark-streaming-flume_2.10-2.1.0.jar

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.flume.FlumeUtilsobject MyFlumeStream {def main(args: Array[String]): Unit = {System.setProperty("hadoop.home.dir", "G:\\bin\\hadoop-2.5.2")Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)val conf = new SparkConf().setAppName("MyFlumeStream").setMaster("local[2]")val ssc = new StreamingContext(conf,Seconds(3))//创建 flume event 从 flume中接收push来的数据 ---> 也是DStream//flume将数据push到了 ip 和 端口中  ip和端口在Flume配置文件中设置val flumeEventDstream = FlumeUtils.createStream(ssc, "192.168.15.131", 1234)val lineDStream = flumeEventDstream.map( e => {new String(e.event.getBody.array)})lineDStream.print()ssc.start()ssc.awaitTermination()}
}

测试:
1、启动Spark Streaming程序。
2、启动Flume。

bin/flume-ng agent -n a4 -f myagent/a4.conf -c conf -Dflume.root.logger=INFO,console

3、拷贝日志文件到/root/training/logs目录。
4、观察输出,采集到数据。

2. 基于Custom Sink的Pull模式

比第一种有更好的健壮性和容错性。生产使用这个方式。

不同于Flume直接将数据推送到Spark Streaming中,第二种模式通过以下条件运行一个正常的Flume Sink。Flume将数据推送到Sink中,并且数据保持buffered状态。Spark Streaming使用一个可靠的Flume接收器和转换器从Sink拉取数据。只要当数据被接收并且被Spark Streaming备份后,转换器才运行成功。这样,与第一种模式相比,保证了很好的健壮性和容错能力。然而,这种模式需要为Flume配置一个正常的Sink。

Flume 配置文件

a1.channels = c1
a1.sinks = k1
a1.sources = r1a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/training/logsa1.channels.c1.type = memory
a1.channels.c1.capacity = 100000
a1.channels.c1.transactionCapacity = 100000a1.sinks.k1.type = org.apache.spark.streaming.flume.sink.SparkSink
a1.sinks.k1.channel = c1
a1.sinks.k1.hostname = 192.168.15.131
a1.sinks.k1.port = 1234#组装source、channel、sink
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

Spark Streaming Full Demo:
注意除了需要使用Flume的lib的jar包以外,还需要
将Spark的jar包拷贝到Flume的lib目录下,spark-streaming-flume_2.10-2.1.0.jar也需要拷贝到Flume的lib目录下,同时加入IDEA工程的classpath。

import org.apache.log4j.Logger
import org.apache.log4j.Level
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.storage.StorageLevelobject FlumeLogPull {def main(args: Array[String]): Unit = {System.setProperty("hadoop.home.dir", "G:\\bin\\hadoop-2.5.2")Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)val conf = new SparkConf().setAppName("FlumeLogPull").setMaster("local[2]")val ssc = new StreamingContext(conf,Seconds(1))val flumeEvent = FlumeUtils.createPollingStream(ssc, "192.168.15.131", 1234, StorageLevel.MEMORY_ONLY_SER)val lineDStream = flumeEvent.map( e => {new String(e.event.getBody.array)})lineDStream.print()ssc.start()ssc.awaitTermination()}
}

测试:
1、启动Flume。

bin/flume-ng agent -n a1 -f myagent/a1.conf -c conf -Dflume.root.logger=INFO,console

2、启动Spark Streaming程序。
3、拷贝日志文件到/root/training/logs目录。
4、观察输出,采集到数据。

这篇关于Spark Streaming(三)—— 高级数据源Flume的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1074030

相关文章

SpringBoot多数据源配置完整指南

《SpringBoot多数据源配置完整指南》在复杂的企业应用中,经常需要连接多个数据库,SpringBoot提供了灵活的多数据源配置方式,以下是详细的实现方案,需要的朋友可以参考下... 目录一、基础多数据源配置1. 添加依赖2. 配置多个数据源3. 配置数据源Bean二、JPA多数据源配置1. 配置主数据

MySQL高级查询之JOIN、子查询、窗口函数实际案例

《MySQL高级查询之JOIN、子查询、窗口函数实际案例》:本文主要介绍MySQL高级查询之JOIN、子查询、窗口函数实际案例的相关资料,JOIN用于多表关联查询,子查询用于数据筛选和过滤,窗口函... 目录前言1. JOIN(连接查询)1.1 内连接(INNER JOIN)1.2 左连接(LEFT JOI

前端高级CSS用法示例详解

《前端高级CSS用法示例详解》在前端开发中,CSS(层叠样式表)不仅是用来控制网页的外观和布局,更是实现复杂交互和动态效果的关键技术之一,随着前端技术的不断发展,CSS的用法也日益丰富和高级,本文将深... 前端高级css用法在前端开发中,CSS(层叠样式表)不仅是用来控制网页的外观和布局,更是实现复杂交

SpringBoot利用dynamic-datasource-spring-boot-starter解决多数据源问题

《SpringBoot利用dynamic-datasource-spring-boot-starter解决多数据源问题》dynamic-datasource-spring-boot-starter是一... 目录概要整体架构构想操作步骤创建数据源切换数据源后续问题小结概要自己闲暇时间想实现一个多租户平台,

kotlin中的行为组件及高级用法

《kotlin中的行为组件及高级用法》Jetpack中的四大行为组件:WorkManager、DataBinding、Coroutines和Lifecycle,分别解决了后台任务调度、数据驱动UI、异... 目录WorkManager工作原理最佳实践Data Binding工作原理进阶技巧Coroutine

深入解析Spring TransactionTemplate 高级用法(示例代码)

《深入解析SpringTransactionTemplate高级用法(示例代码)》TransactionTemplate是Spring框架中一个强大的工具,它允许开发者以编程方式控制事务,通过... 目录1. TransactionTemplate 的核心概念2. 核心接口和类3. TransactionT

SpringBoot操作spark处理hdfs文件的操作方法

《SpringBoot操作spark处理hdfs文件的操作方法》本文介绍了如何使用SpringBoot操作Spark处理HDFS文件,包括导入依赖、配置Spark信息、编写Controller和Ser... 目录SpringBoot操作spark处理hdfs文件1、导入依赖2、配置spark信息3、cont

Python中列表的高级索引技巧分享

《Python中列表的高级索引技巧分享》列表是Python中最常用的数据结构之一,它允许你存储多个元素,并且可以通过索引来访问这些元素,本文将带你深入了解Python列表的高级索引技巧,希望对... 目录1.基本索引2.切片3.负数索引切片4.步长5.多维列表6.列表解析7.切片赋值8.删除元素9.反转列表

Spring Boot实现多数据源连接和切换的解决方案

《SpringBoot实现多数据源连接和切换的解决方案》文章介绍了在SpringBoot中实现多数据源连接和切换的几种方案,并详细描述了一个使用AbstractRoutingDataSource的实... 目录前言一、多数据源配置与切换方案二、实现步骤总结前言在 Spring Boot 中实现多数据源连接

正则表达式高级应用与性能优化记录

《正则表达式高级应用与性能优化记录》本文介绍了正则表达式的高级应用和性能优化技巧,包括文本拆分、合并、XML/HTML解析、数据分析、以及性能优化方法,通过这些技巧,可以更高效地利用正则表达式进行复杂... 目录第6章:正则表达式的高级应用6.1 模式匹配与文本处理6.1.1 文本拆分6.1.2 文本合并6