IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python脚本如何生成Storm拓扑配置

wen 实用脚本 24

本文目录导读：

Python脚本如何生成Storm拓扑配置

  1. 直接生成 YAML/JSON 配置文件（最常用）
  2. 使用模板引擎生成（适合动态配置）
  3. 生成 Java 代码文件（用于实际提交）
  4. 生成 Python 脚本直接提交拓扑（使用 pystorm）
  5. 最佳实践建议
  6. 关键建议：

Python 生成 Storm 拓扑配置通常有以下几种方式，根据你的实际场景选择最合适的方法：

直接生成 YAML/JSON 配置文件（最常用）

import yaml
import json
def generate_storm_topology_yaml():
    config = {
        "name": "my-storm-topology",
        "spouts": [
            {
                "id": "kafka-spout",
                "className": "org.apache.storm.kafka.KafkaSpout",
                "parallelism": 3,
                "config": {
                    "kafka.bootstrap.servers": "localhost:9092",
                    "kafka.topic": "input-topic",
                    "kafka.group.id": "storm-consumer"
                }
            }
        ],
        "bolts": [
            {
                "id": "filter-bolt",
                "className": "com.example.FilterBolt",
                "parallelism": 5,
                "config": {
                    "filter.field": "status",
                    "filter.value": "active"
                }
            },
            {
                "id": "output-bolt",
                "className": "com.example.OutputBolt",
                "parallelism": 2
            }
        ],
        "connections": [
            {"from": "kafka-spout", "to": "filter-bolt", "grouping": "shuffle"},
            {"from": "filter-bolt", "to": "output-bolt", "grouping": "fields", "fields": ["user_id"]}
        ],
        "config": {
            "topology.workers": 3,
            "topology.max.spout.pending": 1000,
            "topology.message.timeout.secs": 300
        }
    }
    # 生成 YAML
    with open('topology.yaml', 'w') as f:
        yaml.dump(config, f, default_flow_style=False)
    # 生成 JSON
    with open('topology.json', 'w') as f:
        json.dump(config, f, indent=2)
    return config
# 生成配置
topology_config = generate_storm_topology_yaml()

使用模板引擎生成（适合动态配置）

from jinja2 import Template
def generate_storm_config_with_template(spout_configs, bolt_configs, env="production"):
    # Jinja2 模板
    template_str = """
name: {{ topology_name }}
spouts:
{% for spout in spouts %}
  - id: {{ spout.id }}
    className: {{ spout.class_name }}
    parallelism: {{ spout.parallelism }}
    config:
      {% for key, value in spout.config.items() %}
      {{ key }}: {{ value }}
      {% endfor %}
{% endfor %}
bolts:
{% for bolt in bolts %}
  - id: {{ bolt.id }}
    className: {{ bolt.class_name }}
    parallelism: {{ bolt.parallelism }}
    config:
      {% for key, value in bolt.config.items() %}
      {{ key }}: {{ value }}
      {% endfor %}
{% endfor %}
connections:
{% for conn in connections %}
  - from: {{ conn.from_id }}
    to: {{ conn.to_id }}
    grouping: {{ conn.grouping }}
    {% if conn.fields %}
    fields: {{ conn.fields }}
    {% endif %}
{% endfor %}
config:
  topology.workers: {{ workers }}
  topology.max.spout.pending: {{ max_pending }}
  topology.message.timeout.secs: {{ timeout }}
  env: {{ environment }}
"""
    template = Template(template_str)
    # 环境特定配置
    env_configs = {
        "development": {"workers": 1, "max_pending": 100, "timeout": 30},
        "staging": {"workers": 3, "max_pending": 500, "timeout": 120},
        "production": {"workers": 10, "max_pending": 5000, "timeout": 300}
    }
    config = env_configs[env]
    config.update({
        "topology_name": "my-topology",
        "spouts": spout_configs,
        "bolts": bolt_configs,
        "connections": generate_connections(spout_configs, bolt_configs),
        "environment": env
    })
    return template.render(**config)
def generate_connections(spouts, bolts):
    connections = []
    # 自动生成连接逻辑
    for spout in spouts:
        for bolt in bolts:
            if bolt.input_type == spout.output_type:
                connections.append({
                    "from_id": spout.id,
                    "to_id": bolt.id,
                    "grouping": "shuffle"
                })
    return connections
# 使用示例
spouts = [
    {
        "id": "kafka-spout",
        "class_name": "org.apache.storm.kafka.KafkaSpout",
        "parallelism": 3,
        "config": {"topic": "input", "bootstrap.servers": "localhost:9092"},
        "output_type": "raw_data"
    }
]
bolts = [
    {
        "id": "filter-bolt",
        "class_name": "com.example.FilterBolt",
        "parallelism": 5,
        "config": {"threshold": 100},
        "input_type": "raw_data"
    }
]
config = generate_storm_config_with_template(spouts, bolts, "production")
with open('topology_generated.yaml', 'w') as f:
    f.write(config)

生成 Java 代码文件（用于实际提交）

def generate_storm_java_topology(topology_name, spouts, bolts):
    java_code = f"""
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.kafka.spout.KafkaSpout;
public class {topology_name}Topology {{
    public static void main(String[] args) throws Exception {{
        TopologyBuilder builder = new TopologyBuilder();
        // 设置 Spouts
"""
    # 添加 Spout 代码
    for spout in spouts:
        java_code += f"""
        builder.setSpout("{spout['id']}", 
            new {spout['class_name']}({spout.get('config_params', '')}), 
            {spout['parallelism']});
"""
    # 添加 Bolt 代码
    java_code += "\n        // 设置 Bolts\n"
    for bolt in bolts:
        java_code += f"""
        builder.setBolt("{bolt['id']}", 
            new {bolt['class_name']}({bolt.get('config_params', '')}), 
            {bolt['parallelism']})
            .{bolt['grouping']}("{bolt['from_spout']}");
"""
    # 添加配置
    java_code += f"""
        Config config = new Config();
        config.setDebug(false);
        config.setNumWorkers({workers_count});
        if (args != null && args.length > 0) {{
            StormSubmitter.submitTopology(args[0], config, builder.createTopology());
        }} else {{
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("{topology_name}", config, builder.createTopology());
            Thread.sleep(10000);
            cluster.shutdown();
        }}
    }}
}}
"""
    return java_code
# 使用示例
spouts = [
    {
        "id": "kafka-spout",
        "class_name": "org.apache.storm.kafka.spout.KafkaSpout",
        "parallelism": 3,
        "config_params": "KafkaSpoutConfig.builder(\"localhost:9092\", \"topic\").build()",
        "grouping": "shuffle",
        "output_field": "value"
    }
]
bolts = [
    {
        "id": "filter-bolt",
        "class_name": "com.example.FilterBolt",
        "parallelism": 2,
        "grouping": "shuffle",
        "from_spout": "kafka-spout"
    }
]
workers_count = 3
java_code = generate_storm_java_topology("MyStormTopology", spouts, bolts)
with open('MyStormTopology.java', 'w') as f:
    f.write(java_code)

生成 Python 脚本直接提交拓扑（使用 pystorm）

import subprocess
import json
def generate_storm_submit_script(topology_name, jar_path, config):
    # 生成提交命令
    storm_cmd = f"""
storm jar {jar_path} org.apache.storm.flux.Flux \\
    --local \\  # 或 --remote 提交到集群
    --sleep 5000 \\
    -R yaml \\
    -filter '{{
        "name": "{topology_name}",
        "config": {json.dumps(config)}
    }}'
"""
    # 或者生成 Flux YAML 配置
    flux_config = {
        "name": topology_name,
        "config": config,
        "components": [],
        "spouts": [
            {
                "id": "kafka-spout",
                "className": "org.apache.storm.kafka.spout.KafkaSpout",
                "constructorArgs": [
                    {"type": "string", "value": "localhost:9092"},
                    {"type": "string", "value": "input-topic"}
                ]
            }
        ],
        "bolts": [
            {
                "id": "process-bolt",
                "className": "com.example.ProcessBolt",
                "parallelism": 2,
                "constructorArgs": [
                    {"type": "string", "value": "param1"}
                ]
            }
        ],
        "streams": [
            {
                "from": "kafka-spout",
                "to": "process-bolt",
                "grouping": {
                    "type": "SHUFFLE"
                }
            }
        ]
    }
    with open(f'flux_{topology_name}.yaml', 'w') as f:
        yaml.dump(flux_config, f)
    return storm_cmd
# 使用示例
config = {
    "topology.workers": 3,
    "topology.max.spout.pending": 5000,
    "topology.debug": False
}
submit_script = generate_storm_submit_script(
    "my-topology", 
    "/path/to/storm-starter-1.2.3.jar",
    config
)
# 保存脚本
with open('submit_topology.sh', 'w') as f:
    f.write("#!/bin/bash\n")
    f.write(submit_script)

最佳实践建议

# 配置文件生成器类
class StormConfigGenerator:
    def __init__(self, topology_name, environment="development"):
        self.topology_name = topology_name
        self.environment = environment
        self.config = {
            "spouts": [],
            "bolts": [],
            "streams": [],
            "components": [],
            "config": self._get_base_config()
        }
    def _get_base_config(self):
        base_config = {
            "topology.workers": 3,
            "topology.max.spout.pending": 1000,
            "topology.message.timeout.secs": 300
        }
        if self.environment == "production":
            base_config.update({
                "topology.workers": 10,
                "topology.max.spout.pending": 50000,
                "topology.debug": False
            })
        elif self.environment == "development":
            base_config.update({
                "topology.workers": 1,
                "topology.max.spout.pending": 10,
                "topology.debug": True
            })
        return base_config
    def add_spout(self, spout_id, class_name, parallelism=1, **kwargs):
        self.config["spouts"].append({
            "id": spout_id,
            "className": class_name,
            "parallelism": parallelism,
            **kwargs
        })
    def add_bolt(self, bolt_id, class_name, parallelism=1, **kwargs):
        self.config["bolts"].append({
            "id": bolt_id,
            "className": class_name,
            "parallelism": parallelism,
            **kwargs
        })
    def add_stream(self, from_id, to_id, grouping="SHUFFLE", **kwargs):
        self.config["streams"].append({
            "from": from_id,
            "to": to_id,
            "grouping": {"type": grouping},
            **kwargs
        })
    def save(self, filename=None):
        if not filename:
            filename = f"topology_{self.topology_name}_{self.environment}.yaml"
        with open(filename, 'w') as f:
            yaml.dump(self.config, f, default_flow_style=False)
        print(f"Configuration saved to {filename}")
        return self.config
# 使用示例
generator = StormConfigGenerator(
    topology_name="data-pipeline",
    environment="production"
)
generator.add_spout(
    "kafka-spout",
    "org.apache.storm.kafka.spout.KafkaSpout",
    parallelism=3,
    constructorArgs=[
        {"type": "string", "value": "localhost:9092"},
        {"type": "string", "value": "data-topic"}
    ]
)
generator.add_bolt(
    "filter-bolt",
    "com.example.DataFilterBolt",
    parallelism=5,
    constructorArgs=[
        {"type": "int", "value": 100}
    ]
)
generator.add_bolt(
    "output-bolt",
    "com.example.OutputBolt",
    parallelism=2
)
generator.add_stream("kafka-spout", "filter-bolt", "SHUFFLE")
generator.add_stream("filter-bolt", "output-bolt", "FIELDS", 
                     fields=["user_id"])
# 生成配置文件
config = generator.save()

关键建议：

  1. 使用 YAML：Storm Flux 原生支持 YAML，是最推荐的方式
  2. 环境分离：开发/测试/生产环境配置分开管理
  3. 版本控制：配置文件纳入 Git 管理
  4. 参数化：使用模板生成动态配置
  5. 验证：生成配置后做基础验证（检查连接完整性等）

选择哪种方式取决于你的部署流程和团队技术栈,YAML + Storm Flux 是官方推荐的主流方式。

上一篇Shell脚本如何实现一键发布策略

下一篇脚本如何批量调整图片色彩查找表

相关文章

抱歉，评论功能暂时关闭!