要从 Kafka 日志中提取 topic 名称及其对应的流量信息，可以编写一个 Python 脚本，通过解析 Kafka 的日志文件，识别与 topic 相关的流量数据。Kafka 日志通常包含与生产者和消费者相关的活动信息，包括 topic 名称、分区、偏移量以及消息大小等[^1]。 以下是一个示例 Python 脚本，用于从 Kafka 日志中提取 topic 名称及其流量： ```python import re from collections import defaultdict # Kafka 日志文件路径 log_file_path = 'kafka_server.log' # 正则表达式用于匹配包含 topic 信息的日志条目 topic_traffic_pattern = re.compile(r'.*Topic:([a-zA-Z0-9_-]+).*Size:([0-9]+).*') # 存储每个 topic 的总流量 topic_traffic = defaultdict(int) # 读取日志文件并解析 with open(log_file_path, 'r') as log_file: for line in log_file: match = topic_traffic_pattern.search(line) if match: topic_name = match.group(1) message_size = int(match.group(2)) topic_traffic[topic_name] += message_size # 输出每个 topic 的总流量 for topic, traffic in topic_traffic.items(): print(f"Topic: {topic}, Total Traffic: {traffic} bytes") ``` ### 脚本说明： 1. **正则表达式**：`topic_traffic_pattern` 用于匹配包含 `Topic` 和 `Size` 字段的日志条目。可以根据实际日志格式调整该正则表达式。 2. **日志解析**：脚本逐行读取日志文件，并使用正则表达式提取 topic 名称和消息大小。 3. **流量统计**：使用 `defaultdict` 对每个 topic 的流量进行累加。 4. **结果输出**：最后输出每个 topic 的名称及其对应的总流量（以字节为单位）。 ### 注意事项： - Kafka 日志格式可能因版本或配置不同而略有差异，因此需要根据实际日志格式调整正则表达式。 - 如果日志中没有直接记录消息大小，可以考虑通过其他方式（如 Kafka 监控工具或 API）获取流量数据。 ### 示例输出： ```plaintext Topic: user_activity, Total Traffic: 12345678 bytes Topic: order_updates, Total Traffic: 98765432 bytes ``` ---