Kafka实战－实时日志统计流程

在《Kafka实战－简单示例》 一文中给大家介绍来Kafka的简单示例，演示了如何编写Kafka的代码去生产数据和消费数据，今天给大家介绍如何去整合一个完整的项目，本篇博客我打 算为大家介绍Flume＋Kafka＋Storm的实时日志统计，由于涉及的内容较多，这里先给大家梳理一个项目的运用这些技术的流程。下面是今天的内容 目录：

Flume Kafka Storm

下面开始今天的内容分享。

2.项目流程

在整合这套方案的时候，项目组也是经过一番讨论，在讨论中，观点很多，有人认为直接使用Storm进行实时处理，去掉Kafka环节；也有认为直接使用Kafka的API去消费，去掉Storm的消费环节等等，但是最终组内还是一致决定使用这套方案，原因有如下几点：

业务模块化 功能组件化

我们认为，Kafka在整个环节中充当的职责应该单一，这项目的整个环节她就是一个中间件，下面用一个图来说明这个原因，如下图所示：

整个项目流程如上图所示，这样划分使得各个业务模块化，功能更加的清晰明了。

Data Collection

负责从各个节点上实时收集用户上报的日志数据，我们选用的是Apache的Flume NG来实现。

Data Access

由于收集的数据的速度和数据处理的速度不一定是一致的，因此，这里添加了一个中间件来做处理，所使用的是Apache的Kafka，关于Kafka集群部署，大家可以参考我写的《Kafka实战－Kafka Cluster》。另外，有一部分数据是流向HDFS分布式文件系统了的，方便于为离线统计业务提供数据源。

Stream Computing

在收集到数据后，我们需要对这些数据做实时处理，所选用的是Apache的Storm。关于Storm的集群搭建部署博客后面补上，较为简单。

Data Output

在使用Storm对数据做处理后，我们需要将处理后的结果做持久化，由于对响应速度要求较高，这里采用Redis＋MySQL来做持久化。整个项目的流程架构图，如下图所示：

3.Flume

Flume是一个分布式的、高可用的海量日志收集、聚合和传输日志收集系统，支持在日志系统中定制各类数据发送方（如：Kafka，HDFS 等），便于收集数据。Flume提供了丰富的日志源收集类型，有：Console、RPC、Text、Tail、Syslog、Exec等数据源的收集， 在我们的日志系统中目前我们所使用的是spooldir方式进行日志文件采集，配置内容信息如下所示：

producer.sources.s.type = spooldir

producer.sources.s.spoolDir = /home/hadoop/dir/logdfs

当然，Flume的数据发送方类型也是多种类型的，有：Console、Text、HDFS、RPC等，这里我们系统所使用的是Kafka中间件来接收，配置内容如下所示：

producer.sinks.r.type = org.apache.flume.plugins.KafkaSink

producer.sinks.r.metadata.broker.list=dn1:9092,dn2:9092,dn3:9092

producer.sinks.r.partition.key=0

producer.sinks.r.partitioner. >

 关于，Flume的详细搭建部署，大家可以参考我写的《高可用Hadoop平台－Flume NG实战图解篇》。这里就不多做赘述了。

4.Kafka

 Kafka是一种提供高吞吐量的分布式发布订阅消息系统，她的特性如下所示：

通过磁盘数据结构提供消息的持久化，这种结构对于即使数据达到TB+级别的消息，存储也能够保持长时间的稳定。

搞吞吐特性使得Kafka即使使用普通的机器硬件，也可以支持每秒数10W的消息。

能够通过Kafka Cluster和Consumer Cluster来Partition消息。

 Kafka的目的是提供一个发布订阅解决方案，他可以处理Consumer网站中的所有流动数据，在网页浏览，搜索以及用户的一些行为，这些动

作是较为关键的因素。这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。对于Hadoop这样的日志数据和离线计算系统，这样的方案是一个

解决实时处理较好的一种方案。

 关于Kafka集群的搭建部署和使用，大家可以参考我写的：《Kafka实战－Kafka Cluster》，这里就不多做赘述了。

5.Storm

 Twitter将Storm开源了，这是一个分布式的、容错的实时计算系统，已被贡献到Apache基金会，下载地址如下所示：

http://storm.apache.org/downloads.html

 Storm的主要特点如下：


简单的编程模型。类似于MapReduce降低了并行批处理复杂性，Storm降低了进行实时处理的复杂性。

可以使用各种编程语言。你可以在Storm之上使用各种编程语言。默认支持Clojure、Java、Ruby和Python。要增加对其他语言的支持，只需实现一个简单的Storm通信协议即可。

容错性。Storm会管理工作进程和节点的故障。

水平扩展。计算是在多个线程、进程和服务器之间并行进行的。

可靠的消息处理。Storm保证每个消息至少能得到一次完整处理。任务失败时，它会负责从消息源重试消息。

快速。系统的设计保证了消息能得到快速的处理，使用ØMQ作为其底层消息队列。

本地模式。Storm有一个本地模式，可以在处理过程中完全模拟Storm集群。这让你可以快速进行开发和单元测试。


 Storm集群由一个主节点和多个工作节点组成。主节点运行了一个名为“Nimbus”的守护进程，用于分配代

码、布置任务及故障检测。每个工作节 

点都运行了一个名为“Supervisor”的守护进程，用于监听工作，开始并终止工作进程。Nimbus和Supervisor都能快速失败，而且是无

 状态的，这样一来它们就变得十分健壮，两者的协调工作是由Apache的ZooKeeper来完成的。

 Storm的术语包括Stream、Spout、Bolt、Task、Worker、Stream 

Grouping和Topology。Stream是被处理的数据。Spout是数据源。Bolt处理数据。Task是运行于Spout或Bolt中的 

线程。Worker是运行这些线程的进程。Stream 

Grouping规定了Bolt接收什么东西作为输入数据。数据可以随机分配（术语为Shuffle），或者根据字段值分配（术语为Fields），或者

广播（术语为All），或者总是发给一个Task（术语为Global），也可以不关心该数据（术语为None），或者由自定义逻辑来决定（术语为 

Direct）。Topology是由Stream Grouping连接起来的Spout和Bolt节点网络。在Storm 

Concepts页面里对这些术语有更详细的描述。

 关于Storm集群的搭建部署，博客在下一篇中更新，到时候会将更新地址附在这里，这里就先不对Storm集群的搭建部署做过多的赘述了。

 这里就是为大家介绍的Flume＋Kafka＋Storm的整体流程，后续会给大家用一个项目案例来实践演示这个流程，包括具体的各个模块的编码实践。今天大家可以先熟悉下实时计算项目的流程开发。

7.结束语

 这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！


  （3）sparkstreaming从kafka接入实时数据流最终实现数据可视化展示 

 1）我们通过kafka与各个业务系统的数据对接，将各系统中的数据实时接到kafka；

2）通过sparkstreaming接入kafka数据流，定义时间窗口和计算窗口大小，业务计算逻辑处理；

3）将结果数据写入到mysql；

4）通过可视化平台接入mysql数据库，这里使用的是NBI大数据可视化构建平台；

5）在平台上通过拖拽式构建各种数据应用，数据展示；


  Kafka监控数据采集 

 上篇文章讲解了运维平台的整体设计，对各个部分并未深入介绍，今天将比较重要的一环——监控数据来源进行讲解。


  Kafka 删除主题流程分析 

 之前有个 Kafka 集群的每个节点的挂载磁盘多达 20+ 个，平均每个磁盘约 1T，每个节点的分区日志被平均分配到这些磁盘中，但由于每个分区的数据不一致，而集群节点 log.retention.bytes 这个参数的默认值是 -1，也就是没有任何限制，因此 Kafka 的日志删除日志依赖 log.retention.hours 参数来删除，因此会出现日志未过期，磁盘写满的情况。

针对该集群双十一会遇到某些挂载磁盘被写满的情况，需要手动对主题进行删除以清空磁盘的操作，现在分析删除主题对集群以及客户端会有什么影响，以及 Kafka 都做了哪些动作。


  探究 | kafka-connector 同步 Elasticsearch速度慢根因分析？ 

 1、kafka同步Elasticsearch的方式

之前博文中也有介绍：

方式一：logstash_input_kafka

方式二：kafka_connector

方式三：spark stream

方式四：java程序读写自己实现