文档章节

基于Flume+Kafka+Spark-Streaming的实时流式处理完整流程

闪电
 闪电
发布于 2016/06/28 12:17
字数 808
阅读 2202
收藏 13

基于Flume+Kafka+Spark-Streaming的实时流式处理完整流程

 

1、环境准备,四台测试服务器

spark集群三台,spark1,spark2,spark3

kafka集群三台,spark1,spark2,spark3

zookeeper集群三台,spark1,spark2,spark3

日志接收服务器, spark1

日志收集服务器,redis (这台机器用来做redis开发的,现在用来做日志收集的测试,主机名就不改了)

 

日志收集流程:

日志收集服务器->日志接收服务器->kafka集群->spark集群处理

说明: 日志收集服务器,在实际生产中很有可能是应用系统服务器,日志接收服务器为大数据服务器中一台,日志通过网络传输到日志接收服务器,再入集群处理。

因为,生产环境中,往往网络只是单向开放给某台服务器的某个端口访问的。

 

Flume版本: apache-flume-1.5.0-cdh5.4.9 ,该版本已经较好地集成了对kafka的支持

 

2、日志收集服务器(汇总端)

配置flume动态收集特定的日志,collect.conf  配置如下:

 

# Name the components on this agent
a1.sources = tailsource-1
a1.sinks = remotesink
a1.channels = memoryChnanel-1

# Describe/configure the source
a1.sources.tailsource-1.type = exec
a1.sources.tailsource-1.command = tail -F /opt/modules/tmpdata/logs/1.log

a1.sources.tailsource-1.channels = memoryChnanel-1

# Describe the sink
a1.sinks.k1.type = logger

# Use a channel which buffers events in memory
a1.channels.memoryChnanel-1.type = memory
a1.channels.memoryChnanel-1.keep-alive = 10
a1.channels.memoryChnanel-1.capacity = 100000
a1.channels.memoryChnanel-1.transactionCapacity = 100000

# Bind the source and sink to the channel
a1.sinks.remotesink.type = avro
a1.sinks.remotesink.hostname = spark1
a1.sinks.remotesink.port = 666
a1.sinks.remotesink.channel = memoryChnanel-1

 

日志实时监控日志后,通过网络avro类型,传输到spark1服务器的666端口上

启动日志收集端脚本:

 

bin/flume-ng agent --conf conf --conf-file conf/collect.conf --name a1 -Dflume.root.logger=INFO,console

 

 

 

 

 

3、日志接收服务器

配置flume实时接收日志,collect.conf  配置如下:

 

#agent section  
producer.sources = s  
producer.channels = c  
producer.sinks = r  
  
#source section  
producer.sources.s.type = avro
producer.sources.s.bind = spark1
producer.sources.s.port = 666

producer.sources.s.channels = c  
  
# Each sink's type must be defined  
producer.sinks.r.type = org.apache.flume.sink.kafka.KafkaSink
producer.sinks.r.topic = mytopic
producer.sinks.r.brokerList = spark1:9092,spark2:9092,spark3:9092
producer.sinks.r.requiredAcks = 1
producer.sinks.r.batchSize = 20
producer.sinks.r.channel = c1
 
#Specify the channel the sink should use  
producer.sinks.r.channel = c  
  
# Each channel's type is defined.  
producer.channels.c.type   = org.apache.flume.channel.kafka.KafkaChannel
producer.channels.c.capacity = 10000
producer.channels.c.transactionCapacity = 1000
producer.channels.c.brokerList=spark1:9092,spark2:9092,spark3:9092
producer.channels.c.topic=channel1
producer.channels.c.zookeeperConnect=spark1:2181,spark2:2181,spark3:2181

 

关键是指定如源为接收网络端口的666来的数据,并输入kafka的集群,需配置好topic及zk的地址

启动接收端脚本:

 

bin/flume-ng agent --conf conf --conf-file conf/receive.conf --name producer -Dflume.root.logger=INFO,console

 

 

 

 

 

4、spark集群处理接收数据

 

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import kafka.serializer.StringDecoder
import scala.collection.immutable.HashMap
import org.apache.log4j.Level
import org.apache.log4j.Logger

/**
 * @author Administrator
 */
object KafkaDataTest {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org.apache.spark").setLevel(Level.WARN);
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.ERROR);

    val conf = new SparkConf().setAppName("stocker").setMaster("local[2]")
    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Seconds(1))

    // Kafka configurations

    val topics = Set("mytopic")

    val brokers = "spark1:9092,spark2:9092,spark3:9092"

    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers, "serializer.class" -> "kafka.serializer.StringEncoder")

    // Create a direct stream
    val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

    val urlClickLogPairsDStream = kafkaStream.flatMap(_._2.split(" ")).map((_, 1))

    val urlClickCountDaysDStream = urlClickLogPairsDStream.reduceByKeyAndWindow(
      (v1: Int, v2: Int) => {
        v1 + v2
      },
      Seconds(60),
      Seconds(5));

    urlClickCountDaysDStream.print();

    ssc.start()
    ssc.awaitTermination()
  }
}


spark-streaming接收到kafka集群后的数据,每5s计算60s内的wordcount值

 

 

5、测试结果

 

往日志中依次追加三次日志

 

spark-streaming处理结果如下:

(hive,1)
(spark,2)
(hadoop,2)
(storm,1)

---------------------------------------

(hive,1)
(spark,3)
(hadoop,3)
(storm,1)

---------------------------------------

(hive,2)
(spark,5)
(hadoop,5)
(storm,2)

与预期一样,充分体现了spark-streaming滑动窗口的特性

本文转载自:http://blog.csdn.net/kwu_ganymede/article/details/50761952

闪电
粉丝 75
博文 392
码字总数 6789
作品 0
海淀
技术主管
私信 提问
加载中

评论(3)

雨季之下
雨季之下
Spark+Kafka实时流机器学习实战
课程观看地址:http://www.xuetuwuyou.com/course/147
课程出自学途无忧网:http://www.xuetuwuyou.com

一、课程使用到的软件及版本
①Spark1.6.2
②kafka0.8.2.1
③centos6.5

二、课程适合人群
适合想学kafka,Spark实时流计算,Spark机器学习的学员

三、课程目标
①学好Kafka,及其整个架构实现原理
②熟练运用Spark机器学习
③熟练掌控Spark与Kafka结合,实现实时流计算

四、课程目录
第1课、spark与kafka的介绍
第2课、spark的集群安装
第3课、Spark RDD函数讲解与实战分析
第4课、Spark 的java操作实现简单程序
第5课、SparkRDD原理详细剖析
第6课、Spark 机器学习,API阅读
第7课、Kafka架构介绍以及集群安装
第8课、Kafka生产者Producer的实战
第9课、Kafka消费者Consumer剖析与实战
第10课、Kafka复杂消费者的详细讲解
第11课、Kafka数据安全,以及Spark Kafka Streaming API
第12课、Spark+Kafka+Mysql整合
第13课、Spark 机器学习ALS设计
第14课、Spark ALS协同过滤java实战
第15课、Spark ALS给用户推荐产品
第16课、Spark机器学习后存储到Mysql
第17课、Spark读取Kafka流构建Als模型
第18课、Spark处理Kafka流构建Als模型
第19课、Spark处理Kafka流实现实时推荐算法
第20课、Spark学习经验总结,spark2与spark1的区别,下期预告


推荐组合学习:《深入浅出Spark机器学习实战(用户行为分析)》
课程观看地址:http://www.xuetuwuyou.com/course/144
闪电
闪电 博主

引用来自“smart152819”的评论

1、spark与kafka的介绍播放
2、spark的集群安装播放
3、Spark RDD函数讲解与实战分析播放
4、Spark 的java操作实现简单程序播放
5、SparkRDD原理详细剖析播放
6、Spark 机器学习,API阅读播放
7、Kafka架构介绍以及集群安装播放
8、Kafka生产者Producer的实战播放
9、Kafka消费者Consumer剖析与实战播放
10、Kafka复杂消费者的详细讲解播放
11、Kafka数据安全,以及Spark Kafka Streaming API播放
12、Spark+Kafka+Mysql整合播放
13、Spark 机器学习ALS设计播放
14、Spark ALS协同过滤java实战播放
15、Spark ALS给用户推荐产品播放
16、Spark机器学习后存储到Mysql播放
17、Spark读取Kafka流构建Als模型播放
18、Spark处理Kafka流构建Als模型 播放
19、Spark处理Kafka流实现实时推荐算法播放
20、Spark学习经验总结,spark2与spark1的区别,下期预告

spark+kafka实时流机器学习实战视频,由夜行侠老师精心录制
http://www.itjoin.org/course/detail/57aa8938b52a8f0ddbd81763
这是广告么????
smart152819
smart152819
1、spark与kafka的介绍播放
2、spark的集群安装播放
3、Spark RDD函数讲解与实战分析播放
4、Spark 的java操作实现简单程序播放
5、SparkRDD原理详细剖析播放
6、Spark 机器学习,API阅读播放
7、Kafka架构介绍以及集群安装播放
8、Kafka生产者Producer的实战播放
9、Kafka消费者Consumer剖析与实战播放
10、Kafka复杂消费者的详细讲解播放
11、Kafka数据安全,以及Spark Kafka Streaming API播放
12、Spark+Kafka+Mysql整合播放
13、Spark 机器学习ALS设计播放
14、Spark ALS协同过滤java实战播放
15、Spark ALS给用户推荐产品播放
16、Spark机器学习后存储到Mysql播放
17、Spark读取Kafka流构建Als模型播放
18、Spark处理Kafka流构建Als模型 播放
19、Spark处理Kafka流实现实时推荐算法播放
20、Spark学习经验总结,spark2与spark1的区别,下期预告

spark+kafka实时流机器学习实战视频,由夜行侠老师精心录制
http://www.itjoin.org/course/detail/57aa8938b52a8f0ddbd81763
spark streaming框架简介

1. spark steaming概述 在《spark 基础(上篇)》中,spark streaming是spark体系中的一个流式处理框架。因此,Spark streaming相对于其他流式处理框架就更有优势,用途更加广泛,它能够与s...

ZPPenny
2017/03/28
0
0
Spark的那些事(一)

Spark是一个快速的集群化的实时计算系统。支持Java, Scala, Python 和R语言的高级API。 一 Spark生态: 1111.png 支持Spark Sql用于sql和结构化数据查询处理;支持MLlib用于机器学习;支持G...

假文艺的真码农
2018/01/02
0
0
通过EMR Spark Streaming实时读取Tablestore数据

本文将介绍如何在E-MapReduce中实时流式的处理Tablestore中的数据。 场景设计 随着互联网的发展,企业中积累的数据越来越多,数据的背后隐藏着巨大的价值,在双十一这样的节日中,电子商务企...

琸然
2019/09/26
0
0
Structured Streaming教程(1) —— 基本概念与使用

近年来,大数据的计算引擎越来越受到关注,spark作为最受欢迎的大数据计算框架,也在不断的学习和完善中。在Spark2.x中,新开放了一个基于DataFrame的无下限的流式处理组件——Structured St...

董黎明
2018/12/07
21
0
基于Kafka的实时计算引擎如何选择?Spark or Flink?

目前实时计算的业务场景越来越多,实时计算引擎技术及生态也越来越成熟。以 Spark 和 Flink 为首的实时计算引擎,成为实时计算场景的重点考虑对象。那么,今天就来聊一聊基于 Kafka 的实时计...

哥不是小萝莉
2019/07/31
0
0

没有更多内容

加载失败,请刷新页面

加载更多

微信公众号开发 (4) 网页授权

一、前言 微信公众号开发 (1) 微信接入认证成为开发者 微信公众号开发 (2) 消息处理 微信公众号开发 (3) 菜单处理 本文将实现 网页授权 获取用户基本信息 网页授权流程 填写网页授权回调域名...

郑清
14分钟前
25
0
号外!号外!百度语音开源库更新了

开源:ReactNative集成百度语音开源库,在这篇文章中作者开源了一个ReactNative集成百度语音合成的组件库,今天通过查看Github上的关于开源react-native-baidu-vtts 的issues,有朋友提到了一...

凌宇之蓝
19分钟前
67
0
【springcloud】之Eureka Server集群搭建

Eureka是微服务架构中的注册中心,专门负责服务的注册与发现 ,里面有一个注册表,保存了各服务所在的机器和端口号 下面我们开始配置 在这里,我配置了两个eureka server 1.新建springboot+...

西元yg
39分钟前
53
0
0120 springboot集成Mybatis和代码生成器

在日常开发中,数据持久技术使用的架子使用频率最高的有3个,即spring-jdbc , spring-jpa, spring-mybatis.详情可以看我之前的一篇文章spring操作数据库的3个架子 . spring-jdbc封装的比较少...

春天springcarter
40分钟前
103
0
re.search和re.match有什么区别?

Python re模块中的search()和match()函数有什么区别? 我已经阅读了文档 ( 当前文档 ),但是我似乎从未记得它。 我一直在查找并重新学习它。 我希望有人会用示例清楚地回答它,以便(也许)...

技术盛宴
40分钟前
44
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部