kafka+flume 实时数据处理

1.监测数据处理技术路线

1.1数据层

2.介绍技术

我们很多人在在使用Flume和kafka时，都会问一句为什么要将Flume和Kafka集成？

一般使用Flume+Kafka架构都是希望完成实时流式的日志处理，如果Flume直接对接实时计算框架，当数据采集速度大于数据处理速度，很容易发生数据堆积或者数据丢失，而kafka可以当做一个消息缓存队列，从广义上理解，把它当做一个数据库，可以存放一段时间的数据。第二、Kafka属于中间件，一个明显的优势就是使各层解耦，使得出错时不会干扰其他组件。因此数据从数据源到flume再到Kafka时，可以做实时计算，可实现数据多分发。

flume

Flume 是一个分布式、可靠、和高可用的海量日志聚合的系统，支持在系统中定制各类数据发送方，通过监控整个文件目录或者某一个特定文件，用于收集数据；同时Flume也提供数据写到各种数据接受方(可定制)的能力，用于转发数据。Flume 的易用性在于通过读取配置文件，可以自动收集日志文件，在大数据处理及各种复杂的情况下，flume 经常被用来作为数据处理的工具。

Source：Flume 搜集日志的方式多种多样，比如可以检测文件夹的变化spool Source，可以监测端口信息 Netcat Source，可以监控某各文件新增的内容 Exec Source等等，通常使用检测文件夹变化的方式来实时收集信息，所以本例中我们也将使用Spool Source。
C

hannel：提供了一层缓冲机制，来实现数据的事务性传输，最大限度保证数据的安全传输。常用的有MemoryChannel：所有的events 被保存在内存中，优点是高吞吐，缺点是容量有限并且Agent 死掉时会丢失内存中的数据；

FileChannel：所有的Events 被保存在文件中，优点是容量较大且死掉时数据可恢复，缺点是速度较慢。因此为了保证Event 在数据流点对点传输中是可靠地，要注意Channel 的选择。目前为了提高速度，我们暂时采用MemoryChannel，之后的目标是实现一个自定义channel—doubleChannel，解决上述的两个痛点问题。

Sink：将数据转发到目的地，或者继续将数据转发到另外一个source，实现接力传输，多层之间通过AVRO Sink来实现。本例中，我们的最终目标是实现实时数据处理，因此实时的采集数据流就把数据发送到Kafka 中。

kafka

Kafka 是由LinkedIn 开发的开源分布式消息系统，主要用于处理LinkedIn 的活跃数据，说白了也就是用户访问日志数据。这些数据主要包括PV、UV、用户行为(登陆、浏览、搜索、分享、点击)、系统运行日志(CPU、内存、磁盘、进程、网络)等方面的数据。这些数据通常以日志的形式进行存储，现有的消息队列系统可以很好的用于日志分析系统对于实时数据的处理，提高日志解析效率。那么说到Kafka，就必须掌握三个原理部分：Producer、Topic、Consumer：

Producer：消息和数据的生产者，向Kafka的一个topic发布消息的过程即为生产过程，在本例中Flume应该是Producer；
Topic：主题，Kafka处理的消息的不同分类(逻辑概念)，可以根据Topic的不同，去区分处理不同的消息。说的更直白一些，Topic就是起到资源隔离的作用，Producer向指定Topic中产生消息，Consumer再从指定的Topic中消费消息。
Consumer：消息和数据的消费者，订阅topic并处理其发布的消息的过程即为消费过程。

4.配置文件

声明基本组件 Source Channel Sink example2.properties

a1.sources = s1
a1.sinks = sk1
a1.channels = c1

配置Source组件,从Socket中接收数据

a1.sources.s1.type = netcat
a1.sources.s1.bind = node0
a1.sources.s1.port = 44444

配置Sink组件,将接收数据输出到对应kafka

#a1.sinks.sk1.type = logger

a1.sinks.sk1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.sk1.kafka.bootstrap.servers = node0:9092
a1.sinks.sk1.kafka.topic = topic01
a1.sinks.sk1.kafka.flumeBatchSize = 20
a1.sinks.sk1.kafka.producer.acks = 1
a1.sinks.sk1.kafka.producer.linger.ms = 1
a1.sinks.sk1.kafka.producer.compression.type = snappy

配置Channel通道，主要负责数据缓冲

a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100

组件间的绑定

a1.sources.s1.channels = c1
a1.sinks.sk1.channel = c1

注意

A netcat-like source that listens on a given port and turns each line of text into an event. Acts like nc -k -l [host] [port]. In other words, it opens a specified port and listens for data. The expectation is that the supplied data is newline separated text. Each line of text is turned into a Flume event and sent via the connected channel.

一个类似于 netcat 的源，它侦听给定的端口并将每一行文本转换为一个事件。作用类似于 nc -k -l [host] [port]。换句话说，它打开一个指定的端口并监听数据。期望提供的数据是换行符分隔的文本。每行文本都变成一个 Flume 事件并通过连接的通道发送。

5.项目演示