课程 7：消息队列与 Hadoop 数据摄取

什么是消息队列？

消息队列是一种软件组件，它临时存储消息（数据包），并允许发送方和接收方应用程序在不直接相互连接的情况下进行交互。发送方（生产者）将消息放入队列，接收方（消费者）在准备好时提取它们。

在分布式系统中的重要性：

• 解耦：生产者和消费者是独立的。生产者不需要知道消费者是谁，或者它当前是否可用。消费者也不需要知道生产者。
• 异步处理：生产者可以发送消息而无需等待消费者处理它们。这使得应用程序能够在负载下保持响应。
• 可伸缩性：多个生产者可以向队列发送消息，多个消费者可以从中处理消息，从而可以轻松扩展系统的不同部分。
• 负载均衡：消息可以在多个消费者之间分发，有助于平衡工作负载。
• 弹性/容错性：如果消费者发生故障，消息仍保留在队列中，一旦消费者恢复或由另一个消费者实例处理即可。

Apache Kafka 简介

Apache Kafka 是一个流行的开源分布式流处理平台。虽然它可以作为传统的消息队列使用，但它也能够处理高吞吐量、容错的实时数据流。

Kafka 核心概念：

• 生产者 (Producers)：发布（发送）记录流（消息）到 Kafka 主题的应用程序。
• 消费者 (Consumers)：订阅（读取和处理）来自 Kafka 主题的记录流的应用程序。消费者可以是“消费者组”的一部分，以分担处理来自某个主题的消息的负载。
• 主题 (Topics)：记录发布到的命名类别或订阅源名称。Kafka 中的主题是分区的，这意味着一个主题可以分割成多个日志（分区），分布在不同的 Kafka 代理服务器上。这允许并行处理。
• 代理 (Brokers)：构成 Kafka 集群的 Kafka 服务器。每个代理存储某些分区的数据，处理来自生产者和消费者的客户端请求，并管理代理之间的数据复制以实现容错。
• ZooKeeper (历史角色)：Kafka 历史上依赖 Apache ZooKeeper 进行集群协调、配置管理和领导者选举。较新版本的 Kafka 通过 KRaft 模式正在减少这种依赖，但许多现有部署仍在使用 ZooKeeper。

捕获数据更改的策略

当应用程序数据发生更改时，我们通常希望捕获这些更改并将它们发送到 Hadoop 进行分析、存档或进一步处理。常用策略包括：

• 应用程序级事件：应用程序代码本身在发生重大数据更改时发出事件（消息）。这通常是最灵活的方法。
• 数据库触发器：数据库中的触发器可以执行发送消息的过程，但这可能会增加数据库的负载。
• 日志抓取/解析：监控应用程序或数据库日志，解析它们以获取相关更改，并将这些更改作为消息发送。
• 变更数据捕获 (CDC)：专门的工具，用于监控数据库事务日志（例如 MySQL 中的 binlog，PostgreSQL 中的 WAL）并将更改流式传输到消息队列。例如：Debezium。

在本课程中，我们将重点关注发送到 Kafka 的应用程序级事件。

概念工作流程：应用程序通过 Kafka 到 HDFS

使用 Kafka 和 Hadoop 的典型数据摄取管道如下所示：

1. 应用程序事件：源应用程序中发生事件（例如，新用户注册、下订单、生成传感器读数）。
2. 生产者：应用程序（或其组件）充当 Kafka 生产者，并将包含事件数据的消息发送到特定的 Kafka 主题。
3. Kafka 主题：消息持久地存储在 Kafka 主题中，分布在 Kafka 代理之间。
4. 消费者：专用的 Kafka 消费者应用程序（或像 Spark Streaming、Flink 这样的流处理框架）订阅该主题。
5. 数据接收器 (HDFS)：消费者处理消息并将数据写入 HDFS。这可能涉及格式化、批处理或简单的文件追加。

简化示例：使用 Kafka 的 Python 生产者/消费者

此示例演示了如何使用 Python 生产者发送消息，并使用 Python 消费者接收它们，然后将其写入本地文件（模拟 HDFS 写入）。

1. Kafka 生产者 (Python 脚本：`producer.py`)

from kafka import KafkaProducer
import json
import time
import random

# Kafka 代理地址
bootstrap_servers = ['localhost:9092']
# Kafka 主题
topic_name = 'app_events'

producer = None
try:
    producer = KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        value_serializer=lambda v: json.dumps(v).encode('utf-8') # 将消息序列化为 JSON
    )
except Exception as e:
    print(f"无法连接到 Kafka 生产者: {e}")
    exit()

print("已连接到 Kafka。正在发送消息...") # "Connected to Kafka. Sending messages..."

try:
    for i in range(10):
        user_id = random.randint(1000, 2000)
        event_type = random.choice(['login', 'click', 'purchase', 'logout'])
        message = {
            'event_id': f'evt_{time.time_ns()}',
            'user_id': user_id,
            'event_type': event_type,
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S')
        }
        
        producer.send(topic_name, value=message)
        print(f"已发送: {message}") # "Sent:"
        time.sleep(1) # 模拟事件之间的一些延迟

except KeyboardInterrupt:
    print("正在停止生产者...") # "Stopping producer..."
finally:
    if producer:
        producer.flush() # 确保所有待处理的消息都已发送
        producer.close()
    print("生产者已关闭。") # "Producer closed."

2. Kafka 消费者 (Python 脚本：`consumer_to_file.py`)

from kafka import KafkaConsumer
import json

# Kafka 代理地址
bootstrap_servers = ['localhost:9092']
# 要订阅的 Kafka 主题
topic_name = 'app_events'
# 消费者组 ID (允许多个消费者分担工作)
group_id = 'hadoop_ingest_group'
# 用于模拟 HDFS 写入的输出文件
output_file = 'data_ingested.txt'

consumer = None
try:
    consumer = KafkaConsumer(
        topic_name,
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset='earliest', # 如果没有存储偏移量，则从最早的消息开始读取
        group_id=group_id,
        value_deserializer=lambda v: json.loads(v.decode('utf-8')) # 反序列化 JSON 消息
    )
except Exception as e:
    print(f"无法连接到 Kafka 消费者: {e}")
    exit()

print(f"已连接到 Kafka。正在从主题 '{topic_name}' 消费消息...") # "Connected to Kafka. Consuming messages from topic..."
print(f"正在将摄取的数据写入 '{output_file}'。按 Ctrl+C 停止。") # "Writing ingested data to ... Press Ctrl+C to stop."

try:
    with open(output_file, 'a') as f: # 以追加模式打开
        for message in consumer:
            # message.value 包含反序列化后的数据
            event_data = message.value
            print(f"已接收: {event_data}") # "Received:"
            
            # 通过将 JSON 行写入本地文件来模拟写入 HDFS
            f.write(json.dumps(event_data) + '\\n')
            f.flush() # 确保立即写入

except KeyboardInterrupt:
    print("正在停止消费者...") # "Stopping consumer..."
finally:
    if consumer:
        consumer.close()
    print("消费者已关闭。") # "Consumer closed."

运行示例：

1. 确保您的 Kafka (和 ZooKeeper) Docker 容器正在运行。
2. 打开一个终端并运行消费者：python consumer_to_file.py
3. 打开另一个终端并运行生产者：python producer.py
4. 您应该会看到生产者发送的消息和消费者接收的消息。消费者会将这些消息写入 `data_ingested.txt`。

与 Docker化的 Hadoop 集成 (概念性)

要将此 Kafka 设置与课程 6 中的 Docker化 Hadoop 集成：

• 网络：Kafka 代理和 Kafka 消费者应用程序最好与您的 Hadoop 集群位于同一 Docker 网络上（例如，在课程 6 的 `docker-compose.yml` 中定义的 `hadoop_network`）。这允许容器使用其服务名称进行通信。
  • 您可以将 Kafka 和 ZooKeeper 服务添加到您的 Hadoop `docker-compose.yml` 中。
  • 或者，将 `hadoop_network` 创建为外部网络，并让 Kafka 的 compose 文件和 Hadoop 的 compose 文件都使用它。
• 消费者部署：
  • Python 消费者脚本可以在其自己的 Docker 容器中运行。此容器需要安装 Python 和 `kafka-python` 库（通过 Dockerfile）。
  • 如果此消费者容器位于 `hadoop_network` 上，并且具有 HDFS 客户端工具/库，则可以通过寻址 NameNode（例如 `hdfs://namenode:9000/...`）直接写入 HDFS。
• 替代方案 (用于模拟的卷挂载)：

  对于此学习阶段，一种更简单、更解耦的方法是：

  1. 在您的主机上（或在通用的 Python 容器中）运行 Python 消费者。
  2. 消费者将数据写入主机上的目录中的 `data_ingested.txt`。
  3. 将此主机目录作为卷挂载到您的 Hadoop NameNode 容器（或任何具有 HDFS 客户端工具的容器）中。
     示例：在您的 Hadoop `docker-compose.yml` 中为 `namenode` 服务添加一个卷：
     volumes:
  - ./ingested_data_on_host:/tmp/ingested_data_from_kafka # 将主机目录映射到容器目录
  4. 然后，定期（或手动）使用 `docker exec namenode hdfs dfs -put /tmp/ingested_data_from_kafka/* /user/myuser/kafka_data/` 将文件移动到 HDFS 中。这模拟了从暂存区域的批量摄取。