公共安全领域 Kafka 应用实践

一、前言

本案例作为大数据框架在公共安全领域应用实践的开篇之作，将从最基础的数据架构体系优化讲起。在接下来的章节里将详细描述Kafka的基本原理、Kafka增强组件以及基于Kafka的Lambda架构的具体应用场景以及相应的研发成果。

Lambda架构由Storm的作者Nathan Marz提出。旨在设计出一个能满足。实时大数据系统关键特性的架构，具有高容错、低延时和可扩展等特。

Lambda架构整合离线计算和实时计算，融合不可变（Immutability，读写分离和隔离 一系列构原则，可集成Hadoop，Kafka，Storm，Spark,HBase等各类大数据组件。大数据系统的关键问题：如何实时地在任意大数据集上进行查询？大数据再加上实时计算，问题的难度比较大。Lambda架构通过分解的三层架构来解决该问题：Batch Layer，Speed Layer和Serving Layer。如下图所示意。

图1.1 Lambda架构图

数据流进入系统后，同时发往Batch Layer和Speed Layer处理。Batch Layer以不可变模型离线存储所有数据集，通过在全体数据集上不断重新计算构建查询所对应的Batch Views。Speed Layer处理增量的实时数据流，不断更新查询所对应的Real time Views。Serving Layer响应用户的查询请求，合并Batch View和Real time View中的结果数据集到最终的数据集。

二、基于Kafka的Lambda架构

2.1 某省大数据平台实践案例

以某省厅大数据建设方案为例，将Kafka作为统一的数据流通道（data pipeline）。Kafka分为地市和省厅两级，地市数据首先经过流式化处理发送到地市的Kafka，经过标准化后，地市Kafka的再汇集到省厅Kafka。

某省大数据平台实践

2.2 引入Kafka的必要性

在大数据系统中，常常会碰到一个问题，整个大数据是由各个子系统组成，数据需要在各个子系统中高性能、低延迟的不停流转。传统的企业消息系统并不是非常适合大规模的数据处理。容易造成日志数据难以收集，容易丢失信息，Oracle实例之间的管道无法供其它系统使用，数据架构易创建难扩展，数据质量差等问题。为了同时搞定在线应用（消息）和离线应用（数据文件，日志），Kafka就出现了。Kafka可以起到两个作用：

• 降低系统组网复杂度。

• 降低编程复杂度，各个子系统不再是相互协商接口，各个子系统类似插口插在插座上，Kafka承担高速数据总线的作用。

传统数据架构

引入Kafka后，可以构建以流为中心数据架构。Kafka是作为一个全局数据管道。每个系统都向这个中心管道发送数据或者从中获取数据。应用程序或流处理程序可以接入管道并创建新的派生流。这些派生流又可以供其它各种系统使用。

以流为中心的数据架构

三、Kafka技术分析

3.1 Kafka的特点

Kafka可以让合适的数据以合适的形式出现在合适的地方。Kafka的做法是提供消息队列，让生产者单往队列的末尾添加数据，让多个消费者从队列里面依次读取数据然后自行处理。

Kafka消息队列

• 分布式系统，易于向外扩展。所有的producer、broker和consumer都会有多个，均为分布式的。无需停机即可扩展机器。

• 提供Pub/Sub方式的海量消息处理。 据了解，Kafka每秒可以生产约25万消息（50 MB），每秒处理55万消息（110 MB）。

• 以高容错的方式存储海量数据流。

• 保证数据流的顺序，处理关键更新。

• 提供消息的长时间存储，将消息持久化到磁盘，因此可用于批量消费，例如ETL，以及实时应用程序。通过将数据持久化到硬盘以及replication防止数据丢失。

• 能够缓存或持久化数据，支持与批处理系统（如Hadoop）的集成。

• 为实时应用程序提供低延时数据传输和处理。

• 支持online和offline的场景。

• 消息被处理的状态是在consumer端维护，而不是由server端维护。当失败时能自动平衡。

3.2 Kafka原理分析

3.2.1 Kafka总体架构

Kafka总体架构

Kafka的整体架构非常简单，是显式分布式架构，producer、broker（kafka）和consumer都可以有多个。Producer，consumer实现Kafka注册的接口，数据从producer发送到broker，broker承担一个中间缓存和分发的作用。broker分发注册到系统中的consumer。broker的作用类似于缓存，即活跃的数据和离线处理系统之间的缓存。客户端和服务器端的通信，是基于简单、高性能且与编程语言无关的TCP协议。

基本概念：

• Topic：特指Kafka处理的消息源（feeds of messages）的不同分类。

• Partition：Topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。

• Message：消息，是通信的基本单位，每个producer可以向一个topic（主题）发布一些消息。

• Producers：消息和数据生产者，向Kafka的一个topic发布消息的过程叫做producers。

• Consumers：消息和数据消费者，订阅topics并处理其发布的消息的过程叫做consumers。

• Broker：缓存代理，Kafka集群中的一台或多台服务器统称为broker。

3.2.2 Kafka关键技术点

3.2.2.1 zero-copy

在Kafka上，有两个原因可能导致低效：一是太多的网络请求，二是过多的字节拷贝。为了提高效率，Kafka把message分成一组一组的，每次请求会把一组message发给相应的consumer。 此外，为了减少字节拷贝，采用了sendfile系统调用。

3.2.2.2 Exactly once message transfer

在Kafka中仅保存了每个consumer已经处理数据的offset。这样有两个好处：一是保存的数据量少；二是当consumer出错时，重新启动consumer处理数据时，只需从最近的offset开始处理数据即可。

3.2.2.3 Push/pull

Producer 向Kafka推（push）数据，consumer 从kafka 拉（pull）数据。

3.2.2.4 负载均衡和容错

Producer和broker之间没有负载均衡机制。broker和consumer之间利用zookeeper进行负载均衡。所有broker和consumer都会在zookeeper中进行注册，且zookeeper会保存他们的一些元数据信息。如果某个broker和consumer发生了变化，所有其他的broker和consumer都会得到通知。

3.2.2.5 分区

Kafka可以将主题划分为多个分区（Partition），会根据分区规则选择把消息均匀的分布到不同的分区中，这样就实现了负载均衡和水平扩展。多个订阅者可以从一个或者多个分区中同时消费数据，以支撑海量数据处理能力。由于消息是以追加到分区中的，多个分区顺序写磁盘的总效率要比随机写内存还要高，是Kafka高吞吐率的重要保证之一。

Kafka分区实现负载均衡，水平拓展，高吞吐率

为了保证数据的可靠性，每个分区节点都会设置一个Leader，以及若干节点当Follower。数据写入分区时，Leader除了自己复制一份，还会将数据复制到每个Follower上。若任一follower挂了，Kafka会再找一个follower从leader获取数据。若Leader挂了，则从Follower中抽取一个当Leader。

Kafka分区实现数据的可靠性

3.3 Kafka的技术选型

3.3.1 Confluent Platform概述

Confluent Platform 是一个流数据平台，能够组织管理来自不同数据源的数据，拥有稳定高效的系统。Confluent Platform 很容易的建立实时数据管道和流应用。通过将多个来源和位置的数据集成到一个中央数据流平台。Confluent Platform简化了连接数据源到Kafka、Kafka构建应用程序，以及安全、监控和管理Kafka的基础设施。

Confluent Platform架构

3.3.2 Kafka Connect

Kafka Connect，可以更方便的创建和管理数据流管道。它为Kafka和其它系统创建规模可扩展的、可信赖的流数据提供了一个简单的模型，通过connectors可以将大数据从其它系统导入到Kafka中，也可以从Kafka中导出到其它系统。Kafka Connect可以将完整的数据库注入到Kafka的Topic中，或者将服务器的系统监控指标注入到Kafka，然后像正常的Kafka流处理机制一样进行数据流处理。而导出工作则是将数据从Kafka Topic中导出到其它数据存储系统、查询系统或者离线分析系统等。

Kafka Connect特性包括：

• Kafka connector通用框架,提供统一的集成API

• 同时支持分布式模式和单机模式

• REST 接口，用来查看和管理Kafka connectors

• 自动化的offset管理，开发人员不必担心错误处理的影响

• 分布式、可扩展

• 流/批处理集成

Kafka connect工作原理

3.4 Kafka端到端审计

采用开源的Chaperone技术框架来实现对kafka的端到端审计。其目标是在数据流经数据管道的每个阶段，能够抓住每个消息，统计一定时间段内的数据量，并尽早准确地检测出数据的丢失、延迟和重复情况。

• 是否有数据丢失？是，那么丢失了多少数据？它们是在数据管道的哪个地方丢失的？

• 端到端的延迟是多少？如果有消息延迟，是从哪里开始的？

• 是否有数据重复？

Chaperone架构

Chaperone架构：AuditLibrary、ChaperoneService、ChaperoneCollector和WebService，它们会收集数据，并进行相关计算，自动检测出丢失和延迟的数据，并展示审计结果。在审计过程中保证每个消息只被审计一次，在层间使用一致性的时间戳。

Chaperone模块审计流程如下：

1. 生成审计消息：ChaperoneService通过定时向特定的Kafka主题生成审计消息来记录状态

2. 审计算法：AuditLibrary实现了审计算法，它会定时收集并打印统计时间窗

3. 获取审计结果：ChaperoneCollector监听特定的Kafka主题，并获取所有的审计消息，存到数据库，生成仪表盘。仪表盘展示：数据的丢失情况、消息的延迟情况、查看每个主题中心的主题状态

4. 准确展示结果：WebService提供了REST接口来查询Chaperone收集到的度量指标。通过这些接口，我们可以准确地计算出数据丢失的数量。

四、Kafka应用成果介绍

基于Kafka的技术特性，Kafka已成熟运用于某省厅的资源服务平台项目，主要用于收集日志、海量数据的微ETL，为各业务系统之间的数据共享提供一个大规模消息处理平台，以及在各地市与省厅之间形成一个数据管道。

结合对Kafka和Kafka插件的深入研究，新德汇大数据研究院自主研发了轻量级的FSP流处理引擎，用于轻便对接流数据，高效处理和实现各类流数据延展应用。

4.1 日志聚合

多个系统之间的日志通过kafka汇聚，提供审计或其他监控系统进行消费。日志聚合一般来说是从服务器上收集日志文件，然后放到一个集中的位置（文件服务器或HDFS）进行处理。然而Kafka忽略掉文件的细节，将其更清晰地抽象成一个个日志或事件的消息流。这就让Kafka处理过程延迟更低，更容易支持多数据源和分布式数据处理。比起以日志为中心的系统比如Scribe或者Flume来说，Kafka提供同样高效的性能和因为复制导致的更高的耐用性保证，以及更低的端到端延迟。

4.2 消息系统

系统之间解耦，通过kafka驱动各业务系统之间的数据共享与业务流程驱动。

比起大多数的消息系统来说，Kafka有更好的吞吐量，内置的分区、冗余及容错性，让Kafka成为了一个很好的大规模消息处理应用的解决方案。消息系统一般吞吐量相对较低，但是需要更小的端到端延时，并常常依赖于Kafka提供的强大的持久性保障。在这个领域，Kafka足以媲美传统消息系统，如ActiveMR或RabbitMQ。

4.3 数据管道

Kafka让集成工作只需连接到一个单独的管道，而无需连接到每个数据生产方与消费方。

Kafka提供数据管道，让多个地市各种类型的数据资源，集成时不需要知道原始数据源的细节，发布数据时也不需要知道哪个应用程序会消费和加载这些数据，增加新系统，也只需要接入现有的Kafka流数据平台就可以。

某省厅Kafka数据管道案例

4.4 ETL流水线

未引入kafka时，数据的ETL过程需生成临时数据库，多次产生落地的文件，耗费内存，而且在再调用临时数据库时，会耗用内存。这样厚重的架构也不具备流数据处理能力。

引入kafka后，实现微ETL。通过Kafka对接流处理引擎，简化ELT流程，细化数据处理层次，低延时获取目标数据。

微ETL优点：

• 无缝衔接流处理引擎，完成数据快速ETL

• kafka构建一个可伸缩的，可靠的数据流通道

• 交互低延迟

• 微ETL实现轻便的数据处理流程

传统ETL与微ETL的对比

4.5 FSP流处理引擎

4.5.1 FSP架构

FSP架构

流处理平台：对流数据，提供核心处理引擎，流采集工具的可配置化管理平台

核心处理引擎：PIPELINEDB允许我们通过sql的方式，对数据流做操作，并把操作结果储存起来；Kafka插件可扩展kafka功能，实现SQL on kafka的各类流数据的延展应用

流采集工具集：Kafkacat实现Kafka与 sqluldr、copy收集的数据的对接，实现流数据的采集

4.5.2 Kafkacat

4.5.2.1 抓取发送消息的工具

Kafkacat是NON JVM TOOL，速度快，轻便，静态编译小于150kb，提供元数据列表展示集群/分区/主题。

Kafkacat工作模式

4.5.2.2 通过kafkacat命令加载数据生成GP外部表

通过Kafkacat实现GP与kafka的数据对接：kafkacat工具根据外部表协议可以获取GP和kafka的数据，并生成外部表，实现数据的并行加载。以外部表的形式实现数据格式错误行的容错处理

Kafkacat 加载GP外部表

五、Kafka延展应用展望

整合NiFi与kafka，并将MiNiFi作为数据采集器布放到对端数据源，形成一条可拓展并流动的流式数据处理生产线。

Kafka与NiFi结合

5.1 NiFi介绍

NiFi是一个易用、强大、可靠的数据处理与分发系统。简单来说，NiFi是用于自动化管理系统之间的数据流。通过与Kafka的对接，提供可视化命令与控制，实现数据流的展示与编辑处理功能，实现数据流的全程追踪。

NiFi特点：

1.可视化命令与控制

基于Web的用户界面，无缝体验设计，监视，控制数据流。

2. 高扩展性

NiFi通过提供自定义类装载器模型，来确保每个扩展组件之间的约束关系被限制在非常有限的程度。因此，在创建扩展组件时，就不用再过多关注其是否会与其他组件产生冲突。数据流处理程序能够以可预测和可重复的模式执行。

3. 数据回压

NiFi提供所有队列数据的缓存，并且在队列达到指定限制或者超时的时候，能够提供数据回压。

4. 高度可配置

数据丢失容错和保证交付，低延迟和高吞吐量，动态优先级，流可以在运行时修改。

5. 安全性

系统间，NiFi可以通过双向SSL进行数据加密。并且可以允许在发送与接收端使用共享密钥，及其他机制对数据流进行加密与解密。

用户与系统间，NiFi允许双向SSL鉴定，并且提供可插入授权模式，因此可以控制用户的登录权限（例如：只读权限、数据流管理者、系统管理员）。

5.2 NiFi实现统一实时采集数据的分布式流平台

数据实时采集器MiNiFi：

• 实现增量数据和流数据的实时采集，而不是传统的定时采集，实现了更细致化的数据获取

• 可支持多种数据源，适用性强

• 实现端到端的数据采集

分布式流平台NiFi：

• 采集而来的数据，形成数据流，并对数据源进行自动记录，索引，跟踪

• 精确控制数据流

• NIFI单节点的性能是每秒处理百兆级数据，搭建NIFI集群可以提升到每秒处理G级别数据

NiFi分布式流平台

作者介绍：

杨刚，现任珠海市新德汇信息技术有限公司副总经理兼大数据研究院院长 15年IT从业经验，长期从事云和大数据的技术研发和实施工作，有深厚的电信、政务、金融等行业背景。