基于Flume+Kafka+Spark

17次阅读
没有评论

行业资讯    
服务器    
云计算    
基于 Flume+Kafka+Spark-Streaming 的实时流式处理过程是怎样的

这篇文章给大家介绍 基于 Flume+Kafka+Spark-Streaming 的实时流式处理过程是怎样的,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。

基于 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-cdh6.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、测试结果

基于 Flume+Kafka+Spark-Streaming 的实时流式处理过程是怎样的

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

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 滑动窗口的特性

关于 基于 Flume+Kafka+Spark-Streaming 的实时流式处理过程是怎样的就分享到这里了,希望以上内容可以对大家有一定的帮助,可以学到更多知识。如果觉得文章不错,可以把它分享出去让更多的人看到。