实用脚本能自动管理Kafka主题吗？

wen 实用脚本 2026-07-21 2

本文目录导读：

实用脚本能自动管理Kafka主题吗？

批量创建主题脚本
自动清理过期主题脚本
主题配置批量修改脚本
Python自动化管理脚本
定时维护脚本（Cron）
YAML配置驱动的脚本
使用建议

是的，可以通过脚本自动管理Kafka主题，以下是一些实用脚本示例,覆盖常见管理场景：

批量创建主题脚本

#!/bin/bash
# create_kafka_topics.sh
KAFKA_HOME="/opt/kafka"
BOOTSTRAP_SERVER="localhost:9092"
TOPICS_FILE="topics.txt"
# topics.txt 格式: topic_name partitions replication_factor
# example:
# user-events 3 2
# order-logs 5 3
# system-metrics 1 1
while IFS=' ' read -r topic partitions replication
do
    echo "Creating topic: $topic with $partitions partitions and $replication replication"
    $KAFKA_HOME/bin/kafka-topics.sh \
        --create \
        --bootstrap-server $BOOTSTRAP_SERVER \
        --topic $topic \
        --partitions $partitions \
        --replication-factor $replication \
        --if-not-exists
    if [ $? -eq 0 ]; then
        echo "✅ Successfully created topic: $topic"
    else
        echo "❌ Failed to create topic: $topic"
    fi
done < "$TOPICS_FILE"

自动清理过期主题脚本

#!/bin/bash
# cleanup_old_topics.sh
KAFKA_HOME="/opt/kafka"
BOOTSTRAP_SERVER="localhost:9092"
RETENTION_DAYS=30
CUTOFF_DATE=$(date -d "$RETENTION_DAYS days ago" +%s)
# 获取所有主题列表
topics=$($KAFKA_HOME/bin/kafka-topics.sh \
    --list \
    --bootstrap-server $BOOTSTRAP_SERVER)
for topic in $topics; do
    # 跳过内部主题
    [[ $topic == __* ]] && continue
    # 获取主题创建时间（需要启用主题元数据）
    create_time=$($KAFKA_HOME/bin/kafka-topics.sh \
        --describe \
        --bootstrap-server $BOOTSTRAP_SERVER \
        --topic $topic \
        | grep -oP 'CreateTime:\K\d+' 2>/dev/null)
    if [ -n "$create_time" ] && [ $create_time -lt $CUTOFF_DATE ]; then
        echo "Deleting old topic: $topic (created: $(date -d @$create_time))"
        $KAFKA_HOME/bin/kafka-topics.sh \
            --delete \
            --bootstrap-server $BOOTSTRAP_SERVER \
            --topic $topic
    fi
done

主题配置批量修改脚本

#!/bin/bash
# update_topic_configs.sh
KAFKA_HOME="/opt/kafka"
BOOTSTRAP_SERVER="localhost:9092"
TAG_PREFIX="env:production"
# 为特定环境的所有主题设置统一的配置
topics=$($KAFKA_HOME/bin/kafka-topics.sh \
    --list \
    --bootstrap-server $BOOTSTRAP_SERVER)
for topic in $topics; do
    # 只处理生产环境主题
    if [[ $topic == $TAG_PREFIX* ]] || [[ $topic == "prod-"* ]]; then
        echo "Updating config for topic: $topic"
        # 设置保留时间和大小
        $KAFKA_HOME/bin/kafka-configs.sh \
            --bootstrap-server $BOOTSTRAP_SERVER \
            --entity-type topics \
            --entity-name $topic \
            --alter \
            --add-config "retention.ms=604800000,retention.bytes=1073741824"
        # 设置压缩类型
        $KAFKA_HOME/bin/kafka-configs.sh \
            --bootstrap-server $BOOTSTRAP_SERVER \
            --entity-type topics \
            --entity-name $topic \
            --alter \
            --add-config "compression.type=snappy"
    fi
done

Python自动化管理脚本

#!/usr/bin/env python3
# kafka_topic_manager.py
import json
import subprocess
import re
from datetime import datetime, timedelta
class KafkaTopicManager:
    def __init__(self, bootstrap_servers="localhost:9092", kafka_home="/opt/kafka"):
        self.bootstrap_servers = bootstrap_servers
        self.kafka_bin = f"{kafka_home}/bin"
    def run_kafka_cmd(self, cmd_type, args):
        """执行Kafka命令"""
        cmd_map = {
            'topics': f"{self.kafka_bin}/kafka-topics.sh",
            'configs': f"{self.kafka_bin}/kafka-configs.sh",
        }
        cmd = [cmd_map[cmd_type]] + args + ["--bootstrap-server", self.bootstrap_servers]
        result = subprocess.run(cmd, capture_output=True, text=True)
        return result.stdout.strip() if result.returncode == 0 else None
    def list_topics(self, exclude_internal=True):
        """列出所有主题"""
        topics = self.run_kafka_cmd('topics', ['--list'])
        if not topics:
            return []
        topic_list = topics.split('\n')
        if exclude_internal:
            topic_list = [t for t in topic_list if not t.startswith('__')]
        return topic_list
    def get_topic_details(self, topic_name):
        """获取主题详细信息"""
        description = self.run_kafka_cmd('topics', ['--describe', '--topic', topic_name])
        if not description:
            return None
        # 解析描述信息
        lines = description.split('\n')
        if not lines:
            return None
        # 提取基本配置
        main_line = lines[0]
        pattern = r'Topic: (\S+).*?PartitionCount: (\d+).*?ReplicationFactor: (\d+)'
        match = re.search(pattern, main_line)
        if match:
            return {
                'name': match.group(1),
                'partitions': int(match.group(2)),
                'replication_factor': int(match.group(3)),
                'configs': self.get_topic_configs(topic_name)
            }
        return None
    def get_topic_configs(self, topic_name):
        """获取主题配置"""
        output = self.run_kafka_cmd('configs', [
            '--describe',
            '--entity-type', 'topics',
            '--entity-name', topic_name,
            '--all'
        ])
        if not output:
            return {}
        configs = {}
        for line in output.split('\n'):
            if '=' in line:
                key, value = line.split('=', 1)
                configs[key.strip()] = value.strip()
        return configs
    def create_topic(self, name, partitions=3, replication=2, configs=None):
        """创建主题"""
        args = ['--create', '--topic', name, 
                '--partitions', str(partitions), 
                '--replication-factor', str(replication)]
        if configs:
            config_str = ','.join(f"{k}={v}" for k, v in configs.items())
            args.extend(['--config', config_str])
        return self.run_kafka_cmd('topics', args)
    def delete_topic(self, name):
        """删除主题"""
        return self.run_kafka_cmd('topics', ['--delete', '--topic', name])
    def update_config(self, topic_name, configs):
        """更新主题配置"""
        config_str = ','.join(f"{k}={v}" for k, v in configs.items())
        return self.run_kafka_cmd('configs', [
            '--alter',
            '--entity-type', 'topics',
            '--entity-name', topic_name,
            '--add-config', config_str
        ])
    def find_topics_by_pattern(self, pattern):
        """根据模式查找主题"""
        import fnmatch
        all_topics = self.list_topics()
        return [t for t in all_topics if fnmatch.fnmatch(t, pattern)]
    def bulk_operation(self, topics, operation, **kwargs):
        """批量操作主题"""
        results = {}
        for topic in topics:
            try:
                if operation == 'delete':
                    results[topic] = self.delete_topic(topic)
                elif operation == 'update_config':
                    results[topic] = self.update_config(topic, kwargs.get('configs', {}))
                elif operation == 'create':
                    results[topic] = self.create_topic(
                        topic, 
                        kwargs.get('partitions', 3), 
                        kwargs.get('replication', 2)
                    )
            except Exception as e:
                results[topic] = f"Error: {str(e)}"
        return results
# 使用示例
if __name__ == "__main__":
    manager = KafkaTopicManager()
    # 1. 列出所有主题
    print("=== 所有主题列表 ===")
    for topic in manager.list_topics():
        details = manager.get_topic_details(topic)
        if details:
            print(f"Topic: {details['name']}, Partitions: {details['partitions']}, Replication: {details['replication_factor']}")
    # 2. 批量创建主题
    print("\n=== 批量创建主题 ===")
    new_topics = {
        'clickstream': {'partitions': 5, 'replication': 3},
        'user-profiles': {'partitions': 3, 'replication': 2}
    }
    for name, config in new_topics.items():
        result = manager.create_topic(name, **config)
        print(f"Create {name}: {result}")
    # 3. 更新配置
    print("\n=== 更新配置 ===")
    config_updates = {'retention.ms': '604800000', 'compression.type': 'gzip'}
    result = manager.update_config('clickstream', config_updates)
    print(f"Update clickstream config: {result}")
    # 4. 按模式查找并操作
    print("\n=== 查找并操作用户相关主题 ===")
    user_topics = manager.find_topics_by_pattern('user-*')
    print(f"找到用户主题: {user_topics}")

定时维护脚本（Cron）

#!/bin/bash
# /etc/cron.d/kafka-topic-maintenance
# 每天凌晨2点执行主题清理
0 2 * * * root /opt/scripts/cleanup_old_topics.sh >> /var/log/kafka-cleanup.log 2>&1
# 每6小时检查一次主题磁盘使用
0 */6 * * * root /opt/scripts/monitor_topic_disk.sh
# 每周一凌晨3点更新所有主题配置
0 3 * * 1 root /opt/scripts/update_topic_configs.sh

YAML配置驱动的脚本

# kafka_topics_config.yaml
topics:
  - name: user-events
    partitions: 3
    replication_factor: 2
    configs:
      retention.ms: 604800000
      compression.type: snappy
      cleanup.policy: delete
      min.insync.replicas: 2
  - name: order-logs
    partitions: 5
    replication_factor: 3
    configs:
      retention.ms: 2592000000
      compression.type: gzip
      cleanup.policy: compact
      delete.retention.ms: 86400000
  - name: system-metrics
    partitions: 1
    replication_factor: 1
    configs:
      retention.ms: 3600000
      retention.bytes: 1073741824

配套的Python解析脚本：

# yaml_to_kafka.py
import yaml
from kafka_topic_manager import KafkaTopicManager
def load_config_and_apply():
    with open('kafka_topics_config.yaml', 'r') as f:
        config = yaml.safe_load(f)
    manager = KafkaTopicManager()
    for topic_config in config['topics']:
        name = topic_config['name']
        existing = manager.get_topic_details(name)
        if existing:
            print(f"Topic {name} already exists, updating config...")
            if topic_config.get('configs'):
                manager.update_config(name, topic_config['configs'])
        else:
            print(f"Creating topic {name}...")
            manager.create_topic(
                name=name,
                partitions=topic_config['partitions'],
                replication=topic_config['replication_factor'],
                configs=topic_config.get('configs')
            )
if __name__ == "__main__":
    load_config_and_apply()

使用建议

权限管理：生产环境使用Kafka ACL控制脚本权限
错误处理：添加重试机制和错误日志
版本兼容：注意不同Kafka版本的命令差异
安全考虑：敏感信息（密码、密钥）使用环境变量或密钥管理服务
测试环境：先在测试环境验证脚本效果

这些脚本可以根据你的具体需求进行调整,实现Kafka主题的自动化管理。