Flume 安装

Apache Flume 是一个分布式、可靠、高可用的海量日志聚合系统，支持在系统中定制各类数据发送方，用于收集数据；同时，Flume 可以对数据的简单处理，并写到各种数据接收方。

特点

Flume 的数据流由事件（Event）贯穿始终。事件是 Flume 的基本数据单位，它携带日志数据（字节数组形式）并且携带有头信息，这些 Event 由 Agent 外部的 Source 生成，当 Source 捕获事件后会进行特定的格式化，然后 Source 会把事件推入（单个或多个）Channel 中。可以把 Channel 看作是一个缓冲区，它将保存事件直到 Sink 处理完该事件。Sink 负责持久化日志或者把事件推向另一个 Source。

  • Flume 的可靠性
  • 当节点出现故障时，日志能够被传送到其他节点上而不会丢失。Flume 提供了三种级别的可靠性保障、从强到弱依次分别为：end - to - end（收到数据 agent 首先将 event 写到磁盘上，当数据传送成功后，再删除；如果数据发送失败，可以重新发送）；Store on failure（这也是 scribe 采用的策略，当数据接收方 crash 时，将数据写到本地，待恢复后，继续发送）；Besteffort（数据发送到接收方后，不会进行确认）。
  • Flume 的可恢复性
  • 还是靠 Channel。推荐使用 FileChannel，事件持久化在本地文件系统里（性能较差）。

概念

  • Client：Client 生产数据，运行在一个独立的线程。
  • Event：一个数据单元，消息头和消息体组成。（Events 可以是日志记录、avro 对象等。）
  • Flow：Event 从源点到达目的点的迁移的抽象。
  • Agent：一个独立的 Flume 进程，包含组件 Source、Channel、Sink。（Agent 使用 JVM 运行 Flume。每台机器运行一个 Agent，但是可以在一个 Agent 中包含多个 Sources 和 Sinks。）
  • Source：数据收集组件。（Source 从 Client 收集数据，传递给 Channel）
  • Channel：中转 Event 的一个临时存储，保存由 Source 组件传递过来的 Event。（Channel 连接 Sources 和 Sinks，这个有点像一个队列。）
  • Sink：从 Channel 中读取并移除 Event，将 Event 传递到 FlowPipeline 中的下一个 Agent（如果有的话）（Sink 从 Channel 收集数据，运行在一个独立线程。）

Source

Source 是数据的收集端，负责将数据捕获后进行特殊的格式化，将数据封装到事件（Event）里，然后将事件推入 Channel 中。

Flume 提供了各种 Source 的实现，包括 Avro Source、Exce Source、Spooling Directory Source、NetCat Source、Syslog Source、Syslog TCP Source、Syslog UDP Source、HTTP Source、HDFS Source、etc。如果内置的 Source 无法满足需要，Flume 还支持自定义 Source。

Flum-Source

Sink

Flume Sink 取出 Channel 中的数据，进行相应的存储文件系统，数据库，或者提交到远程服务器。

Flume 也提供了各种 Sink 的实现，包括 HDFS sink、Logger sink、Avro sink、File Roll sink、Null sink、HBase sink、etc。

Flume Sink 在设置存储数据时，可以向文件系统中，数据库中，Hadoop 中存储数据，在日志数据较少时，可以将数据存储在文件系统中，并且设定一定的时间间隔保存数据。在日志数据较多时，可以将相应的日志数据存储到 Hadoop 中，便于日后进行相应的数据分析。

Flume-Sink

Flume NG 的体系结构

Flume 运行的核心是 Agent。Flume 以 Agent 为最小的独立运行单位。一个 Agent 就是一个 JVM。它是一个完整的数据收集工具，含有三个核心组件：Source、Channel、Sink。通过这些组件，Event 可以从一个地方流向另一个地方。

Flum-NG 体系结构

Flume 安装

  • 上传解压（『caroly02』）

  • tar xf apache-flume-1.6.0-bin.tar.gz -C /opt/caroly/
cd /opt/caroly
mv apache-flume-1.6.0-bin/ flume

  • 修改配置文件

  • cd flume/conf/

  • mv flume-env.sh.template flume-env.sh

  • vi flume-env.sh

    • change 22: 
export JAVA_HOME=/usr/java/jdk1.8.0_251-amd64

  • vi fk.conf

    • a1.sources = r1
a1.sinks = k1
a1.channels = c1

# Describe/configure the source
a1.sources.r1.type = avro
a1.sources.r1.bind = caroly02
a1.sources.r1.port = 41414

# Describe the sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = testflume
a1.sinks.k1.brokerList = caroly02:9092,caroly03:9092,caroly04:9092
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.batchSize = 20

# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000000
a1.channels.c1.transactionCapacity = 10000

# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

  • 启动

  • 需先启动 Zookeeper 和 Kafka

  • bin/flume-ng agent -n a1 -c conf -f conf/fk.conf -Dflume.root.logger=DEBUG,console