Flume - Kafka日志平台整合

1. Flume介绍

Flume是Cloudera提供的一个高可用的,高可靠的,分布式的海量日志采集、聚合和传输的系统,Flume支持在日志系统中定制各类数据发送方,用于收集数据;同时,Flume提供对数据进行简单处理,并写到各种数据接受方(可定制)的能力。

  • agent agent本身是一个Java进程,运行在日志收集节点—所谓日志收集节点就是服务器节点。 agent里面包含3个核心的组件:source—->channel—–>sink,类似生产者、仓库、消费者的架构。
  • source source组件是专门用来收集数据的,可以处理各种类型、各种格式的日志数据,包括avro、thrift、exec、jms、spooling directory、netcat、sequence generator、syslog、http、legacy、自定义。
  • channel source组件把数据收集来以后,临时存放在channel中,即channel组件在agent中是专门用来存放临时数据的——对采集到的数据进行简单的缓存,可以存放在memory、jdbc、file等等。
  • sink sink组件是用于把数据发送到目的地的组件,目的地包括hdfs、logger、avro、thrift、ipc、file、null、Hbase、solr、自定义。
  • event 将传输的数据进行封装,是flume传输数据的基本单位,如果是文本文件,通常是一行记录,event也是事务的基本单位。event从source,流向channel,再到sink,本身为一个字节数组,并可携带headers(头信息)信息。event代表着一个数据的最小完整单元,从外部数据源来,向外部的目的地去。

2. Kafka Channel && Kafka Sink

2.1 Kafka channel

Kafka channel可以应用在多样的场景中:

  1. Flume source and sink: 可以为event提供一个高可靠性和高可用的channel;
  2. Flume source and interceptor but no sink: 其他应用可以将Fluem event写入kafka topic中;
  3. With Flume sink, but no source: 提供低延迟、高容错的方式将Fluem event从kafka中写入其他sink,例如:HDFS,HBase或者Solr。
  • Kafka Channel配置 加粗部分为必填属性。

Property Name

Default

Description

type

The component type name, needs to be org.apache.flume.channel.kafka.KafkaChannel

kafka.bootstrap.servers

List of brokers in the Kafka cluster used by the channel This can be a partial list of brokers, but we recommend at least two for HA. The format is comma separated list of hostname:port

kafka.topic

flume-channel

Kafka topic which the channel will use

kafka.consumer.group.id

flume

Consumer group ID the channel uses to register with Kafka. Multiple channels must use the same topic and group to ensure that when one agent fails another can get the data Note that having non-channel consumers with the same ID can lead to data loss.

parseAsFlumeEvent

true

Expecting Avro datums with FlumeEvent schema in the channel. This should be true if Flume source is writing to the channel and false if other producers are writing into the topic that the channel is using. Flume source messages to Kafka can be parsed outside of Flume by using org.apache.flume.source.avro.AvroFlumeEvent provided by the flume-ng-sdk artifact

migrateZookeeperOffsets

true

When no Kafka stored offset is found, look up the offsets in Zookeeper and commit them to Kafka. This should be true to support seamless Kafka client migration from older versions of Flume. Once migrated this can be set to false, though that should generally not be required. If no Zookeeper offset is found the kafka.consumer.auto.offset.reset configuration defines how offsets are handled.

pollTimeout

500

The amount of time(in milliseconds) to wait in the “poll()” call of the consumer. https://kafka.apache.org/090/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#poll(long)

defaultPartitionId

Specifies a Kafka partition ID (integer) for all events in this channel to be sent to, unless overriden by partitionIdHeader. By default, if this property is not set, events will be distributed by the Kafka Producer’s partitioner - including by key if specified (or by a partitioner specified by kafka.partitioner.class).

partitionIdHeader

When set, the producer will take the value of the field named using the value of this property from the event header and send the message to the specified partition of the topic. If the value represents an invalid partition the event will not be accepted into the channel. If the header value is present then this setting overrides defaultPartitionId.

kafka.consumer.auto.offset.reset

latest

What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g. because that data has been deleted): earliest: automatically reset the offset to the earliest offset latest: automatically reset the offset to the latest offset none: throw exception to the consumer if no previous offset is found for the consumer’s group anything else: throw exception to the consumer.

kafka.producer.security.protocol

PLAINTEXT

Set to SASL_PLAINTEXT, SASL_SSL or SSL if writing to Kafka using some level of security. See below for additional info on secure setup.

kafka.consumer.security.protocol

PLAINTEXT

Same as kafka.producer.security.protocol but for reading/consuming from Kafka.

more producer/consumer security props

If using SASL_PLAINTEXT, SASL_SSL or SSL refer to Kafka security for additional properties that need to be set on producer/consumer.

2.2 Kafka Sink

Flume 支持将数据发布到一个kafka topic。目前支持Kafka 0.9.x版本。

  • KafkaSink 配置 加粗部分为必填配置

Property

Name

Default Description

type

Must be set to org.apache.flume.sink.kafka.KafkaSink

kafka.bootstrap.servers

List of brokers Kafka-Sink will connect to, to get the list of topic partitions This can be a partial list of brokers, but we recommend at least two for HA. The format is comma separated list of hostname:port

kafka.topic

default-flume-topic

The topic in Kafka to which the messages will be published. If this parameter is configured, messages will be published to this topic. If the event header contains a “topic” field, the event will be published to that topic overriding the topic configured here. Arbitrary header substitution is supported, eg. %{header} is replaced with value of event header named “header”. (If using the substitution, it is recommended to set “auto.create.topics.enable” property of Kafka broker to true.)

flumeBatchSize

100

How many messages to process in one batch. Larger batches improve throughput while adding latency.

kafka.producer.acks

1

How many replicas must acknowledge a message before its considered successfully written. Accepted values are 0 (Never wait for acknowledgement), 1 (wait for leader only), -1 (wait for all replicas) Set this to -1 to avoid data loss in some cases of leader failure.

useFlumeEventFormat

false

By default events are put as bytes onto the Kafka topic directly from the event body. Set to true to store events as the Flume Avro binary format. Used in conjunction with the same property on the KafkaSource or with the parseAsFlumeEvent property on the Kafka Channel this will preserve any Flume headers for the producing side.

defaultPartitionId

Specifies a Kafka partition ID (integer) for all events in this channel to be sent to, unless overriden by partitionIdHeader. By default, if this property is not set, events will be distributed by the Kafka Producer’s partitioner - including by key if specified (or by a partitioner specified by kafka.partitioner.class).

partitionIdHeader

When set, the sink will take the value of the field named using the value of this property from the event header and send the message to the specified partition of the topic. If the value represents an invalid partition, an EventDeliveryException will be thrown. If the header value is present then this setting overrides defaultPartitionId.

allowTopicOverride

true

When set, the sink will allow a message to be produced into a topic specified by the topicHeader property (if provided).

topicHeader

topic

When set in conjunction with allowTopicOverride will produce a message into the value of the header named using the value of this property. Care should be taken when using in conjunction with the Kafka Source topicHeader property to avoid creating a loopback.

kafka.producer.security.protocol

PLAINTEXT

Set to SASL_PLAINTEXT, SASL_SSL or SSL if writing to Kafka using some level of security. See below for additional info on secure setup.

more producer security props

If using SASL_PLAINTEXT, SASL_SSL or SSL refer to Kafka security for additional properties that need to be set on producer.

Other Kafka Producer Properties

These properties are used to configure the Kafka Producer. Any producer property supported by Kafka can be used. The only requirement is to prepend the property name with the prefix kafka.producer. For example: kafka.producer.linger.ms

3. Flume - Kafka配置示例

切换到flume/conf目录下,编辑配置文件:

agent.sources = s1
agent.channels = c1
agent.sinks = k1

# Source Config
agent.sources.s1.type = spooldir
agent.sources.s1.channels = c1
agent.sources.s1.bind = 192.168.100.105
agent.sources.s1.port = 9696
agent.sources.s1.includePattern = *.log
agent.sources.s1.spoolDir = /home/usr/tomcat-test/logs

# Sink Config
## 输出到kafka
agent.sinks.s1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.s1.channel = c1
agent.sinks.s1.topic = test_tomcat_logs
agent.sinks.s1.serializer.class = kafka.serializer.StringEncoder
agent.sinks.s1.brokerList = 192.168.100.105:9092


# Channel Config
agent.channels.c1.type = memory
agent.channels.c1.keep-alive = 10
agent.channels.c1.capacity = 65535

很明显,由配置文件可以了解到:

  1. 我们需要读取目录:/home/usr/tomcat-test/logs下日志文件;
  2. flume连接到kafka的地址是 192.168.100.105:9092,注意不要配置出错了;
  3. flume会将采集后的内容输出到Kafka topic 为test_tomcat_logs,所以我们启动zk,kafka后需要打开一个终端消费topic kafkatest的内容。这样就可以看到flume与kafka之间开始工作了。

4. 运行

运行flume直接切换到flume目录执行以下命令即可:

$ bin/flume-ng agent --conf conf --conf-file example.conf --name a1 -Dflume.root.logger=INFO,console

参考资料: [1] Flume Doc: http://flume.apache.org/FlumeUserGuide.html#kafka-channel

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏小樱的经验随笔

php实现文件上传

存档: upload1.html(单文件上传) 1 <html> 2 <head> 3 <title>单个文件上传</title>...

7566
来自专栏Java成神之路

hbase_异常_04_util.FSUtils: Waiting for dfs to exit safe mode...

可等Hadoop退出安全模式后再执行HBase命令,或者手动退出Hadoop的安全模式

1222
来自专栏琦小虾的Binary

CMake学习笔记(三)——以笔者的Robosub竞赛为例

CMake学习笔记(三)——以笔者的Robosub竞赛为例 继笔者认真学习了CMake语法之后,便开始尝试自己用CMake将以前用Qt写的软件框架程序改编为CM...

34210
来自专栏Hongten

Hibernate中使用memcached

Hibernate-memcached是CacheProvider接口的memcached实现

1802
来自专栏一个会写诗的程序员的博客

8.1 Spring Boot集成Groovy混合Java开发小结

本章节我们使用SpringBoot集成Groovy混合Java开发一个极简的RestAPI。 数据库使用mysql,ORM层使用mybatis,模板引擎使用fr...

1852
来自专栏互联网技术栈

Spring Boot 2.0 Release Notes 中文版

由于1.5版本是Spring Boot的主要版本,所有升级现有的应用程序可能会比平常更复杂一点。我们制定了专门的迁移指南来帮助您升级现有的Spring Boot...

1492
来自专栏SpringBoot

spring+tomcat7 + websocket + sock.js消息推送

使用STOMP 的目的,目前还有浏览器不支持websocket ,所有用了STOMP 

2321
来自专栏杨建荣的学习笔记

内核参数导致的备库宕机分析 (一)r7笔记第23天

在前几天搭建好备库之后,因为同步文件着实花了些时间,首先配置备库能够正常接收归档,然后内核参数也基本没有设置,简单使用脚本算出一个 Hugepage的值,就直接...

3897
来自专栏潇涧技术专栏

Gradle Plugin for Android Development User Guide 1

Gradle Plugin for Android Development User Guide (1)

853
来自专栏学习力

《Java从入门到放弃》框架入门篇:springMVC数据传递 (二)

1784

扫码关注云+社区

领取腾讯云代金券