项目五 动态网页访问日志数据采集
Flume 采集官网访问日志、Kafka 消费日志：Source / Channel / Sink 三段式链路与拦截器。
本项目学习目标
- 能说清 Flume 的 Agent 结构与三种组件
- 能写采集配置并启动、验证有数据
- 能用 Kafka 建 topic、生产与消费消息
- 能判断日志采集链路的常见故障点
本项目任务导览
动态网页访问日志数据采集
使用 Flume 采集某官网访问日志数据，并使用 Kafka 消费 Flume 采集到的日志数据，构建 Flume + Kafka 日志采集架构。
- 项目导言与学习目标
- 任务 5-1 使用 Flume 采集某官网访问日志数据
- 任务 5-2 使用 Kafka 消费 Flume 中某官网访问日志数据
- 知识拓展与实例案例（3 例）
日志是网站运营决策的重要数据来源。
日志中通常包含了大量的用户访问网站的行为信息，比如页面响应时间、消耗的流量、访问时间、停留时间、是否访问成功等，网站运营者可通过分析这些信息，对网站运营做出一定的决策。
了解什么是日志采集； 熟悉日志数据采集的方法； 掌握 Flume 进行数据采集的方法； 精通使用 Kafka 进行消息订阅发布的方法。
具备使用 Flume 拦截器过滤数据的能力； 具备熟悉 Kafka 的脚本操作的能力； 具备掌握 Flume 采集日志文件的能力； 具备掌握使用 Kafka 进行消息发布订阅的能力； 具备精通 Flume + Kafka 架构实现数据采集的能力。
具备事业心和责任心； 具有艰苦奋斗的精神和务实作风； 具有健康的体魄，良好的身体素质； 具备责任心。
任务 5-1 使用 Flume 采集某官网访问日志数据
任务技能：Flume 简介 → 数据采集配置 → Flume 启动 → 拦截器配置。
一、Flume 简介
Flume 是一个由 Cloudera 开发的分布式、可靠和高可用的海量日志收集和传输系统，具有基于流数据的简单灵活的架构。它具有可调整的可靠性机制以及故障转移和恢复机制，具有健壮性和容错性。
Flume 由三层架构组成，分别为 agent、collector 和 storage。
| 组件 | 说明 |
|---|---|
| Source（采集组件） | 用于与数据源进行对接，获取数据。Source 支持的数据源包括 Avro、Thrift、exec、JMS、spooling directory、netcat、sequence generator、syslog、http、legacy、自定义类型。 |
| Channel（通道） | 是一个缓存区，用于链接 Source 和 Sink 组件，缓存 Source 写入的 Event。 |
| Sink（下沉组件） | 负责取出 Channel 中的消息数据，将 Channel 中的 Event 数据发送到文件存储系统或服务器等。 |
Flume 日志收集中常见的场景是将客户端大量生成的日志数据发送到存储子系统或一个消费者代理，即多个数据源汇入一个目的地。
Flume 不仅能够同时采集多个服务器的数据到同一个服务器中，还能够将数据采集并分别保存到不同服务器中，即一个数据源分发到多个目的地。
二、Flume 数据采集配置
Flume 实现数据采集的方式非常简单，只需编辑相应的配置文件即可完成特定数据的采集与存储，配置文件中需要对数据源（source）、通道（channel）和接收器（sink）进行配置。
用于监听 Avro 端口并接收来自外部 Avro 客户端的事件流。当两个 Flume agent 的 Avro Sink 和 Avro Source 对应时，可组成分层集合拓扑结构。
| 属性 | 描述 |
|---|---|
| channels | 指定通道的名称 |
| type | 组件类型名称，值为 avro |
| bind | 监听的主机名或 IP 地址 |
| port | 监听的端口 |
Exec Source 的配置是指设置一个 Linux 或 Unix 命令，通过该命令不断输出数据，当命令进程退出时 Exec Source 也会退出，后续不会产生下一步数据。Exec Source 和其他异步源的问题在于，如果将事件放入 Channel 失败，无法保证数据的完整性。
| 属性 | 描述 |
|---|---|
| channels | 指定通道的名称 |
| type | 组件类型名称，值为 exec |
| command | 要执行的命令 |
# 配置一个agent，agent的名称可自定义
# 指定agent的sources（如r1）、channels（如c1）
a1.sources = r1
a1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /var/log/secure
a1.sources.r1.channels = c1Spooling Directory Source 能够通过磁盘中的目录摄取数据，当监控的指定目录中出现新文件时，会将文件作为 source 进行处理，一旦文件被放到"自动收集"目录中后，便不能修改和重命名。
| 属性 | 描述 |
|---|---|
| channels | 指定通道的名称 |
| type | 组件类型名称，值为 spooldir |
| spoolDir | 读取文件的目录 |
| fileHeader | 是否添加存储绝对路径文件名的文件头 |
# 配置一个agent，agent的名称可自定义
# 指定agent的sources（如r1）、channels（如c1）
a1.sources = r1
a1.channels = c1
a1.sources.r1.type = spooldir
a1.sources.r1.channels = c1
a1.sources.r1.spoolDir = /var/log/apache/flumeSpool
a1.sources.r1.fileHeader = trueFlume 能够作为 Kafka 的消费者，从 Kafka 中读取消息，当有多个 kafka 源在运行时，可以使用相同的 Consumer Group 以便读取一组唯一的主题分区。
| 属性 | 描述 |
|---|---|
| channels | 指定通道的名称 |
| type | 组件类型名称，值为 org.apache.flume.source.kafka.KafkaSource |
| kafka.bootstrap.servers | Kafka 集群中的 broker 列表 |
| kafka.topics | Kafka 消费者将从中读取消息的以逗号分隔的主题列表 |
| kafka.topics.regex | 定义源订阅的主题集的正则表达式，此属性的优先级高于 kafka.topics |
| batchSize | 一批写入 Channel 的最大消息数 |
| batchDurationMillis | 将批次写入通道之前的最长时间（毫秒），只要达到大小和时间中的第一个，就会写入批次 |
| kafka.consumer.group.id | 消费者组的唯一标识 |
监听指定端口并将每一行文本转换为一个事件，与命令 nc -k -l [host] [port] 类似。接收的数据是以换行符分隔的文本，每一行文本都会变成一个 Flume 事件并通过连接的通道发送。
| 属性 | 描述 |
|---|---|
| channels | 指定通道的名称 |
| type | 组件类型名称，值为 netcat |
| bind | 监听的主机 IP 地址 |
| port | 监听的主机端口号 |
# 配置一个agent，agent的名称可自定义
# 指定agent的sources（如r1）、channels（如c1）
a1.sources = r1
a1.channels = c1
a1.sources.r1.type = netcat
a1.sources.r1.bind = 0.0.0.0
a1.sources.r1.port = 6666
a1.sources.r1.channels = c1使用内存作为源与接收器之间的保留区称之为内存通道，内存速度比磁盘快数倍，所以数据传输的速度也会随之加快。
| 属性 | 描述 |
|---|---|
| type | 组件类型名称，需要设置为 memory |
| capacity | channel 里存放的最大 event 数，默认 100 |
| transactionCapacity | Channel 每次提交的 Event 数量，默认 100 |
| byteCapacityBufferPercentage | 定义 byteCapacity 和通道中所有事件的估计总大小之间的缓冲区百分比 |
| byteCapacity | 设置通道中所有事件在内存中的大小总和 |
使用磁盘作为源与接收器之间的保留区称之为文件通道，由于文件通道使用的是磁盘，所以会比内存通道速度慢。文件通道优点在于当遇到断电、硬件故障等能够导致 Flume 采集中断的问题时，重启 Flume 即可不会造成数据丢失。
| 属性 | 描述 |
|---|---|
| type | 组件类型名称，需要设置为 file |
| checkpointDir | 存储检查点文件的目录 |
| dataDirs | 用于存储日志文件，在不同磁盘上使用多个目录可以提高文件通道性能 |
此接收器将数据写入 Hadoop 分布式文件系统（HDFS），支持创建文本和序列文件，使用此接收器需要安装 Hadoop，以便 Flume 与 HDFS 集群进行通信。
| 属性 | 默认值 | 描述 |
|---|---|---|
| channel | - | 指定通道的名称 |
| type | - | 组件类型名称，需要为 hdfs |
| hdfs.path | - | 写入 hdfs 的路径（必选） |
| hdfs.filePrefix | FlumeData | 文件名前缀 |
| hdfs.fileSuffix | - | 文件名后缀 |
| hdfs.inUsePrefix | - | 用于 flume 主动写入的临时文件的前缀 |
| hdfs.inUseSuffix | .tmp | 用于 flume 主动写入的临时文件的后缀 |
| hdfs.rollInterval | 30 | 按时间生成 HDFS 文件，单位：秒 |
| hdfs.rollSize | 1024 | 触发滚动的文件大小，以字节为单位（0: 从不根据文件大小进行滚动） |
| hdfs.rollCount | 10 | 按事件数量生成新文件 |
| hdfs.idleTimeout | 0 | 关闭非活动文件的超时（0 = 禁用自动关闭空闲文件） |
| hdfs.fileType | SequenceFile | 文件格式，包括：SequenceFile、DataStream、CompressedStream |
| hdfs.writeFormat | Writable | 写文件的格式，包含：Text、Writable |
| hdfs.round | false | 是否对时间戳四舍五入 |
| hdfs.roundValue | 1 | 将其四舍五入到最高倍数 |
| hdfs.roundUnit | second | 向下取整的单位，可选值包括 second、minute 或 hour |
| hdfs.minBlockReplicas | - | 设置 HDFS 块的副本数，默认采用 Hadoop 的配置 |
| hdfs.useLocalTimeStamp | False | 在替换转义序列时使用本地时间（而不是事件标头中的时间戳） |
HDFS Sink 能够根据时间、数据大小、事件数定期滚动文件（关闭当前文件并创建新文件），或按时间戳、事件起源的机器等属性对数据进行存储/分区。此时在 hdfs.path 设置中会包含格式转义序列，这些转义序列将被 HDFS 接收器替换以生成目录/文件名来存储事件。
| 转义序列 | 描述 |
|---|---|
| %{host} | 主机名 |
| %t | 以毫秒为单位的 Unix 时间 |
| %a | 语言环境的简短工作日名称（周一、周二…） |
| %A | 语言环境的完整工作日名称（星期一、星期二…） |
| %b | 语言环境的短月份名称（Jan、Feb…） |
| %B | 语言环境的长月份名称（一月、二月…） |
| %c | 语言环境的日期和时间（2005 年 3 月 3 日星期四 23:05:25） |
| %d | 一个月中的某天（01） |
| %e | 没有填充的月份中的某一天（1） |
| %D | 日期；与 %m/%d/%y 相同 |
| %H | 小时，24 小时制（00…23） |
| %I | 小时，12 小时制 |
| %j | 一年中的一天（001..366） |
| %k | 小时（0..23） |
| %m | 月（01..12） |
| %n | 简短月份（1..12） |
| %M | 分钟（00..59） |
| %p | 语言环境相当于 am 或 pm |
| %s | 自 1970-01-01 00:00:00 UTC 以来的秒数 |
| %S | 秒（00..59） |
| %y | 年份的最后两位数字（00..99） |
| %Y | 年（2022） |
Hive Sink 接收器能够将包含特定分隔符的文件或 JSON 数据的事件流传输到 Hive 表中。由于事件是基于 Hive 事务编写的，所以一旦 Flume 将一组事件提交到 Hive 后，在 Hive 表中可立即查询。需要注意的是使用 Hive Sink 时需要将 Hive 表创建为事务表，并开始 Hive 的事务。
| 属性 | 默认值 | 描述 |
|---|---|---|
| channel | - | 指定通道的名称 |
| type | - | 组件类型名称，需要为 hive |
| hive.metastore | - | Hive Metastore URI（例如 thrift://abcom:9083） |
| hive.database | - | Hive 数据库名称 |
| hive.table | - | Hive 中的表名称 |
| hive.partition | - | 使用逗号分隔的分区值列表，标识要写入的分区 |
| heartBeatInterval | 240 | 设置以秒为单位的心跳检测时间间隔，设置为 0 表示禁用心跳检测 |
| autoCreatePartitions | true | Flume 将自动创建必要的 Hive 分区 |
| batchSize | 15000 | 在单个 Hive 事务中写入 Hive 的最大事件数 |
| maxOpenConnections | 500 | 打开链接的最大数量，如果超过此数量，则关闭最近最少使用的连接 |
| callTimeout | 10000 | Hive 和 HDFS I/O 操作的超时时间，单位为秒 |
| serializer | - | 序列化器，负责从事件中解析出字段并映射到配表中的列 |
| roundUnit | minute | 向下取整的单位，可选值包括 second、minute 或 hour |
| roundValue | 1 | 向下舍入到此的最高倍数（在使用 hive.roundUnit 配置的单位中），小于当前时间 |
| timeZone | LocalTime | 用于解析分区中的转义序列的时区名称，例如 America/Los_Angeles |
| useLocalTimeStamp | False | 在替换转义序列时使用本地时间（而不是事件标头中的时间戳） |
该接收器与 Avro Source 可构成分层架构。发送到此接收器的 Flume 事件将转换为 Avro 事件并发送到配置的主机名/端口。
| 属性 | 默认值 | 描述 |
|---|---|---|
| channel | - | 指定通道的名称 |
| type | - | 组件类型名称，需要为 avro |
| hostname | - | 监听的主机名或 IP 地址 |
| port | - | 监听的端口号 |
该接收器能够将事件存储在本地文件系统上。
| 属性 | 默认值 | 描述 |
|---|---|---|
| channel | - | 指定通道的名称 |
| type | - | 组件类型名称，需要为 file_roll |
| sink.directory | - | 将存储文件的目录 |
| sink.rollInterval | 30 | 每 30 秒滚动一次文件。指定 0 将禁用滚动并导致将所有事件写入单个文件 |
三、Flume 启动
启动 Flume 代理使用名为 flume-ng 的 shell 脚本，该脚本位于 Flume 发行版的 bin 目录中。启动时需要在命令中指定代理名称、配置目录以及配置文件，并且设置的代理名称要和配置文件中的代理名称一致。
| 参数 | 描述 |
|---|---|
| --name | 指定 Agent 的名称（必选） |
| --conf | 指定配置文件所在目录 |
| --conf-file | 指定配置文件 |
# 启动 Flume agent 进行数据采集
[root@master flume-code]# cd /usr/local/flume/bin
[root@master bin]# ./flume-ng agent --conf conf --conf-file /usr/local/inspur/code/flume-code/example.conf --name a1单独启动另一个终端，使用 telnet 向 4444 端口向 Flume 发送事件，启动成功后输入"Hello world!"：
[root@master ~]# telnet localhost 4444四、Flume 拦截器配置
Flume 能够在数据采集的过程中对事件进行删除或修改，可理解为对事件进行过滤，这些功能是靠拦截器完成的。Flume 拦截器包括 5 类：时间拦截器、主机拦截器、静态拦截器、搜索和替换拦截器、正则表达式过滤拦截器。
时间拦截器将处理事件的时间（以毫秒为单位）插入到事件头中，并根据时间戳将数据写入到不同文件中。当不使用任何拦截器时，Flume 接收到的只有 message。
| 属性 | 默认值 | 描述 |
|---|---|---|
| type | - | 组件类型名称，值为 timestamp |
| headerName | timestamp | 生成的时间戳的标头的名称 |
| preserveExisting | false | 如果时间戳已经存在，是否应该保留，值为 true 或 false |
a1.sources = r1
a1.channels = c1
a1.sources.r1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = timestamp该拦截器向事件中添加包含当前 Flume 代理的主机名或 IP 地址，主要表现形式为在 HDFS 中显示以 Flume 主机 IP 地址作为前缀的文件名。
| 属性 | 默认值 | 描述 |
|---|---|---|
| type | - | 组件类型名称，值为 host |
| preserveExisting | false | 如果主机头已经存在是否保留，值为 true 或 false |
| useIP | true | 如果为 true，请使用 IP 地址，否则使用主机名 |
| hostHeader | host | 要使用的文件头 |
a1.sources = r1
a1.channels = c1
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = host静态拦截器能够将具有静态值的静态事件头附加到所有事件，同一个静态拦截器中不能够设置多个事件头，但可以设置多个静态拦截器。
| 属性 | 默认值 | 描述 |
|---|---|---|
| type | - | 组件类型名称，值为 static |
| preserveExisting | true | 配置的标头已经存在，是否应该保留，值为 true 或 false |
| key | key | 事件头名称 |
| value | value | 静态值 |
a1.sources = r1
a1.channels = c1
a1.sources.r1.channels = c1
a1.sources.r1.type = avro
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = static
a1.sources.r1.interceptors.i1.key = datacenter
a1.sources.r1.interceptors.i1.value = NEW_YORK该拦截器提供了基于 Java 正则表达式的字符串的搜索和替换功能。
| 属性 | 默认值 | 描述 |
|---|---|---|
| type | - | 组件类型名称，值为 search_replace |
| searchPattern | true | 要搜索和替换的模式 |
| replaceString | key | 替换字符串 |
| charset | value | 事件主体的字符集，默认情况下假定为 UTF-8 |
a1.sources.r1.interceptors = search-replace
a1.sources.r1.interceptors.search-replace.type = search_replace
# 删除事件正文中的前导字母数字字符
a1.sources.r1.interceptors.search-replace.searchPattern = ^[A-Za-z0-9_]+
a1.sources.r1.interceptors.search-replace.replaceString =该拦截器通过将事件主体解释为文本，并将文本与配置的正则表达式匹配完成事件的过滤。
| 属性 | 默认值 | 描述 |
|---|---|---|
| type | - | 组件类型名称，值为 regex_filter |
| regex | "*" | 用于匹配事件的正则表达式 |
| excludeEvents | false | 如果为 true，则正则表达式确定要排除的事件，否则正则表达式确定要包含的事件 |
a1.sources.s1.interceptors = i1
a1.sources.s1.interceptors.i1.type = regex_filter
a1.sources.s1.interceptors.i1.regex = (\\d):(\\d):(\\d)任务 5-2 使用 Kafka 消费 Flume 中某官网访问日志数据
Kafka 作为消息中间件，消费 Flume 采集到的日志数据。
一、Kafka 简介
Kafka 是由 Apache 软件基金会开发的一个开源流处理平台，是一个快速、可扩展的、分布式的、分区的、可复制的基于 zookeeper 协调的分布式日志系统，用于 web/nginx 日志、访问日志、消息服务等，Linkedin 于 2010 年将 Kafka 贡献给了 Apache 基金会并成为顶级开源项目。
- 将 zookeeper 安装包上传到"/usr/local"目录下并解压重命名为 zookeeper；
- 将模板文件"zoo_sample.cfg"文件复制重命名为"zoo.cfg"；
- 进入 zookeeper 安装目录的"bin"目录，启动 zookeeper 并查看进程；
- 将 Kafka 安装包上传到"/usr/local"目录下解压并重命名为 kafka；
- 在 Kafka 的"/bin"目录下后台启动 Kafka 进程。
二、Kafka 常用脚本
Kafka 为开发人员提供了众多脚本对 Kafka 进行管理和相关的操作，其中常用的脚本包含 Topic 脚本、生产者脚本、消费者脚本等。
通过该脚本文件能够完成创建主题、查看主题、修改主题、删除主题等操作。
| 参数 | 描述 |
|---|---|
| --alter | 修改主题 |
| --config <String: name=value> | 创建或修改主题时，用于设置主题级别的参数 |
| --create | 创建主题 |
| --delete | 删除主题 |
| --describe | 查看主题的详细信息 |
| --disable-rack-aware | 创建主题时不考虑机架信息 |
| --help | 打印帮助信息 |
| --if-exists | 修改或删除主题时使用，只有当主题存在时才会执行动作 |
| --if-not-exists | 创建主题时使用，只有主题不存在时才会执行动作 |
| --list | 列出所有可用的主题 |
| --partitions 分区数 | 创建主题或增加分区时指定的分区数 |
| --replica-assignment 分配方案 | 手工指定分区副本分配方案 |
| --replication-factor 副本数 | 创建主题时指定副本数 |
| --topic 主题名称 | 指定主题名称 |
| --topics-with-overrides | 使用 describe 查看主题信息时，只展示包含覆盖配置的主题 |
| --unavailable-partitions | 使用 describe 查看主题信息时，只展示包含没有 leader 副本的分区 |
| --under-replicated-partitions | 使用 describe 查看主题信息时，只展示包含失效副本的分区 |
| --bootstrap-server | 链接的服务器地址和端口（必需） |
该脚本能够生产创建消息并将消息发送到消费者。
| 参数 | 描述 |
|---|---|
| --bootstrap-server | 要连接的服务器（必需，除非指定 --broker-list） |
| --topic | （必需）topic 名称 |
| --batch-size | 单个批处理中发送的消息数（默认值：16384） |
| --compression-codec | 压缩编解码器，可选值"none"、"gzip"、"snappy"、"lz4"或"zstd"。如果未指定值，则默认为"gzip" |
| --max-block-ms | 在发送请求期间，生产者将阻止的最长时间（默认值：60000） |
| --max-memory-bytes | 生产者用来缓冲等待发送到服务器的总内存（默认值：33554432） |
| --max-partition-memory-bytes | 为分区分配的缓冲区大小（默认值：16384） |
| --message-send-max-retries | 最大的重试发送次数（默认值：3） |
| --metadata-expiry-ms | 强制更新元数据的时间阈值(ms)（默认值：300000） |
| --producer-property | 将自定义属性传递给生成器的机制 |
| --producer.config | 生产者配置属性文件（--producer-property 优先于此配置），配置文件完整路径 |
| --property | 自定义消息读取器 |
| --request-required-acks | 生产者请求的确认方式 |
| --request-timeout-ms | 生产者请求的确认超时时间 |
| --retry-backoff-ms | 生产者重试前，刷新元数据的等待时间阈值 |
| --socket-buffer-size | TCP 接收缓冲大小 |
| --timeout | 消息排队异步等待处理的时间阈值 |
| --sync | 同步发送消息 |
该脚本接收并消费生产者发送的消息。
| 参数 | 描述 |
|---|---|
| --group | 指定消费者所属组的 ID |
| --topic | 被消费的 topic |
| --partition | 指定分区；除非指定 --offset，否则从分区结束（latest）开始消费 |
| --offset | 执行消费的起始 offset 位置；默认值：latest；/latest /earliest /偏移量 |
| --whitelist | 正则表达式匹配 topic（--topic 就不用指定了）；匹配到的所有 topic 都会消费；用了这个参数，--partition --offset 等就不能使用了 |
| --consumer-property | 将用户定义的属性以 key=value 的形式传递给使用者 |
| --consumer.config | 消费者配置属性文件（--consumer-property 优先于此配置） |
| --property | 初始化消息格式化程序的属性 |
| --from-beginning | 从存在的最早消息开始，而不是从最新消息开始；注意如果配置了客户端名称并且之前消费过，那就不会从头消费了 |
| --max-messages | 消费的最大数据量，若不指定，则持续消费下去 |
| --skip-message-on-error | 如果处理消息时出错，请跳过它而不是暂停 |
| --isolation-level | 设置为 read_committed 以过滤掉未提交的事务性消息；设置为 read_uncommitted 以读取所有消息；默认值：read_uncommitted |
三、Kafka API
生产者 API 顾名思义，就是指应用程序发送数据流到 Kafka 集群中的 topic 中，供消费者消费使用。生产者是线程安全的，跨线程共享单个生产者实例通常比拥有多个实例更快。
| 参数 | 描述 |
|---|---|
| bootstrap_server | 要连接的服务器（必需，除非指定 --broker-list） |
| batch_size | 单个批处理中发送的消息数（默认值：16384） |
| compression_type | 生产者生成的所有数据的压缩类型，有效值为"gzip"、"snappy"、"lz4"或"无" |
| max_block_ms | 在发送请求期间，生产者将阻止的最长时间（默认值：60000） |
| buffer_memory | 生产者用来缓冲等待发送到服务器的总内存（默认值：33554432） |
| metadata_max_age_ms | 强制更新元数据的时间阈值(ms)（默认值：300000） |
| request_timeout_ms | 生产者请求的确认超时时间，默认值：30000 |
| retry_backoff_ms | 生产者重试前，刷新元数据的等待时间阈值，默认值：100 |
消费者 API 主要用于接收生产者集群发出的消息，并且消费者不是线程安全的，不应使用跨线程共享。
| 参数 | 描述 |
|---|---|
| topics (str) | 要订阅的可选主题列表 |
| bootstrap_servers | 要连接的服务器 |
| client_id (str) | 此客户端的名称，此字符串在每个请求中传递给服务器，可用于标识与此客户端对应的特定服务器端日志条目 |
| group_id | 消费者组 ID |
通过消费者 API 成功接收生产者消息后，会返回一个消息集合，每条消息会包含 topic、消息所在的分区、消息所在分区的位置、消息的内容、消息的 key，对应获取方法如下表。
| 方法 | 描述 |
|---|---|
| partition | 消息所在的分区 |
| offset | 消息所在分区的位置 |
| key | 消息的 key |
| value | 消息的内容 |
在"/usr/local/inspur/code/pykafka"目录中编写 python 代码，使用 API 创建消费者接收"test-topic"中的消息：
# vim consumer-demo.py 代码如下
consumer = KafkaConsumer('test-topic', bootstrap_servers='localhost:9092')
for message in consumer:
print("topic = %s, partition = %d, offset = %d, value = %s"
% (message.topic, message.partition, message.offset, message.value))
# 运行
[root@master pykafka]# python consumer-demo.py知识拓展与实例案例
课堂讲解用：3 个可运行实例 + 知识拓展卡片。
知识拓展
实例 1：Flume 监控目录实时采集到 HDFS
目标：用 Spooling Directory Source 监控日志目录，把新增文件实时采集到 HDFS。
# flume-spool-hdfs.conf
# 组件定义：sources / channels / sinks
agent.sources = s1
agent.channels = c1
agent.sinks = k1
# Source：监控目录 /usr/local/inspur/data
agent.sources.s1.type = spooldir
agent.sources.s1.spoolDir = /usr/local/inspur/data
agent.sources.s1.fileSuffix = .COMPLETED
agent.sources.s1.deletePolicy = immediate
# Channel：File 通道保证可靠性
agent.channels.c1.type = file
agent.channels.c1.checkpointDir = /usr/local/flume/checkpoint
agent.channels.c1.dataDirs = /usr/local/flume/data
# Sink：写入 HDFS，按日期分目录
agent.sinks.k1.type = hdfs
agent.sinks.k1.hdfs.path = /flume/events/%Y%m%d
agent.sinks.k1.hdfs.filePrefix = event
agent.sinks.k1.hdfs.fileType = DataStream
# 绑定关系
agent.sources.s1.channels = c1
agent.sinks.k1.channel = c1
# 启动
[root@master flume]# bin/flume-ng agent -n agent -c conf -f flume-spool-hdfs.conf| 配置项 | 白话解释 |
|---|---|
| agent.sources = s1 / channels = c1 / sinks = k1 | 先声明三个"组件"并起名：数据从哪来（sources=s1）、临时放哪（channels=c1）、送到哪去（sinks=k1）。 |
| agent.sources.s1.type = spooldir | source 类型设为 spooldir（监控目录）：Flume 会一直盯着某个文件夹，有新文件就自动采集。 |
| agent.sources.s1.spoolDir = /usr/local/inspur/data | 指定要监控的文件夹路径，日志文件放进去即被采集。 |
| agent.sources.s1.fileSuffix = .COMPLETED | 采集完的文件会被改名、加上 .COMPLETED 后缀，防止重复采集同一个文件。 |
| agent.channels.c1.type = file | 通道类型用 file：数据先写进本地磁盘暂存。相比 memory 通道，file 更可靠，程序重启也不丢数据。 |
| agent.sinks.k1.type = hdfs | sink 类型设为 hdfs：把数据最终写入 Hadoop 分布式文件系统。 |
| hdfs.path = /flume/events/%Y%m%d | HDFS 上的保存路径；%Y%m%d 是日期占位符（如 20230210），每天自动建一个文件夹，方便按天归档。 |
| bin/flume-ng agent -n agent -c conf -f 文件 | 启动命令：-n 指定组件名，-c 指定配置目录，-f 指定刚才编写的配置文件。 |
实例 2：Flume 采集日志并转发到 Kafka
目标：用 Exec Source 实时读取日志文件追加内容，经 Memory Channel 转发到 Kafka 的 test-topic。
# flume-exec-kafka.conf
agent.sources = s1
agent.channels = c1
agent.sinks = k1
# Source：tail -F 跟踪文件追加内容
agent.sources.s1.type = exec
agent.sources.s1.command = tail -F /usr/local/inspur/data/access.log
agent.sources.s1.restart = true
# Channel：Memory 通道，容量 10000 事件
agent.channels.c1.type = memory
agent.channels.c1.capacity = 10000
agent.channels.c1.transactionCapacity = 1000
# Sink：输出到 Kafka
agent.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.k1.kafka.bootstrap.servers = localhost:9092
agent.sinks.k1.kafka.topic = test-topic
agent.sinks.k1.flumeBatchSize = 5
agent.sources.s1.channels = c1
agent.sinks.k1.channel = c1| 配置项 | 白话解释 |
|---|---|
| agent.sources.s1.type = exec | source 类型设为 exec（执行命令）：Flume 会运行一条命令，从命令输出中采集数据。 |
| agent.sources.s1.command = tail -F .../access.log | 要执行的命令是 tail -F 文件：持续跟踪日志文件，有新内容追加就立即输出一行——实时采集的关键。 |
| agent.sources.s1.restart = true | 命令意外退出后自动重启，保证采集进程不中断。 |
| agent.channels.c1.type = memory | 通道用 memory（内存）：速度快，但数据只存在内存里，进程崩溃会丢失。适合对可靠性要求不高的场景。 |
| capacity = 10000 / transactionCapacity = 1000 | 通道最多暂存 10000 条事件；每次事务最多处理 1000 条。相当于"缓冲区大小"。 |
| agent.sinks.k1.type = ...KafkaSink | sink 类型设为 Kafka 输出插件：把数据直接发给 Kafka。 |
| kafka.bootstrap.servers = localhost:9092 | Kafka 服务器的地址和端口，告诉 Flume"消息要送到哪里"。 |
| kafka.topic = test-topic | 消息发到名为 test-topic 的主题（可以理解为 Kafka 里的"队列/分类"）。 |
实例 3：Kafka 命令行完整收发流程
目标：完整演示创建 Topic、启动生产者、启动消费者、发送与接收消息的过程。
# 1. 启动 Kafka（依赖 ZooKeeper）
[root@master kafka]# bin/zookeeper-server-start.sh config/zookeeper.properties &
[root@master kafka]# bin/kafka-server-start.sh config/server.properties &
# 2. 创建 Topic：1 个分区 1 个副本
[root@master kafka]# bin/kafka-topics.sh --create --topic test-topic \
--bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
# 3. 启动消费者（终端 A）
[root@master kafka]# bin/kafka-console-consumer.sh --topic test-topic \
--bootstrap-server localhost:9092
# 4. 启动生产者并发送消息（终端 B）
[root@master kafka]# bin/kafka-console-producer.sh --topic test-topic \
--bootstrap-server localhost:9092
> hello kafka
> 第一条日志消息
> ^C
# 5. 消费者终端立即收到：
hello kafka
第一条日志消息| 命令 | 白话解释 |
|---|---|
| bin/zookeeper-server-start.sh config/zookeeper.properties | 启动 ZooKeeper——Kafka 的"管理员"，负责记录集群元信息。Kafka 依赖它才能运行。 |
| bin/kafka-server-start.sh config/server.properties | 启动 Kafka 服务本身（Broker），server.properties 是它的配置文件。 |
| kafka-topics.sh --create --topic test-topic --partitions 1 --replication-factor 1 | 创建一个主题 test-topic：partitions=1 表示分成 1 个分区；replication-factor=1 表示 1 个副本。Topic 可以理解为 Kafka 里的"消息队列"。 |
| kafka-console-consumer.sh --topic test-topic --bootstrap-server localhost:9092 | 启动"控制台消费者"：在终端 A 里订阅 test-topic，消息一来就打印。--bootstrap-server 指定 Kafka 地址。 |
| kafka-console-producer.sh --topic test-topic --bootstrap-server localhost:9092 | 启动"控制台生产者"：在终端 B 里输入消息，回车即发送到 test-topic。 |
| > hello kafka / > 第一条日志消息 | 生产者输入的内容。> 是控制台提示符，不是命令的一部分。 |
| ^C | Ctrl+C 的表示：结束生产者程序。 |