风暴拓扑配置答案

【问题标题】：Storm topology configuration风暴拓扑配置
【发布时间】：2013-08-06 08:20:01
【问题描述】：

如何为风暴拓扑提供自定义配置？例如，如果我构建了一个连接到 MySQL 集群的拓扑，并且我希望能够在不重新编译的情况下更改需要连接的服务器，我该怎么做？我的偏好是使用配置文件，但我担心文件本身没有部署到集群，因此它不会运行（除非我对集群如何工作的理解有缺陷）。到目前为止，我所见过的在运行时将配置选项传递给风暴拓扑的唯一方法是通过命令行参数，但是当您获得大量参数时，这会很麻烦。

确实有一个想法是利用 shell 脚本将文件读入变量并将该变量的内容作为字符串传递给拓扑，但如果可能的话，我想要一些更简洁的东西。

有没有其他人遇到过这种情况？如果有，你是怎么解决的？

编辑：

似乎需要提供更多说明。我的场景是我有一个拓扑，我希望能够在不同的环境中部署，而无需重新编译它。通常，我会创建一个包含数据库连接参数之类的配置文件并将其传入。我想知道如何在 Storm 中执行类似的操作。

【问题讨论】：

我想一个公平的问题是问为什么不重新编译它？构建 jar 的时间应该不会很大。
我在将要部署到的系统上没有编译器。例如，与任何数据库的连接都会有所不同，因此我必须能够更改该部分配置而无需重新编译。我也不是负责部署的人，所以它需要简单。我目前的解决方案是利用 Properties 对象并从文件中读取配置。然后，我从中填充storm Config 对象，从而使所有选项都可用于所有螺栓。我只是将螺栓的“名称”添加到简单隔离的选项中。
除非我误解了你，否则我们使用 Flux 来实现。 storm.apache.org/releases/2.0.0-SNAPSHOT/flux.html您可以将特定于环境的配置放在单独的文件中并将它们添加到包含部分？

标签： apache-storm

【解决方案1】：

实际上最适合您的是将配置存储在可变键值存储（s3、redis 等）中，然后将其拉入以配置您随后使用的数据库连接（我假设您已经在计划限制您与数据库交谈的频率，以便获取此配置的开销不是什么大问题）。这种设计允许您即时更改数据库连接，甚至无需重新部署拓扑。

【讨论】：

我已经考虑过这种方法，但我仍然需要一种方法来告诉我的拓扑服务器所在的位置。
为什么？部署不能作为 spout 或 bolt 计算一部分的拓扑时需要进行哪些配置？
您可以将配置保存在键值存储、dB、zookeeper 等中，并在启动时在 cmdline 或资源中包含的配置文件中指定服务器（名称、端口等）配置，以及与 jar 一起分发到每个节点，我们为我们的几个配置执行此操作。

【解决方案2】：

这个想法是，当您构建拓扑时，您会创建 spout 和 bolt 的实例（除其他外），这些实例会被序列化并分发到集群中的正确位置。如果你想配置一个 spout 或 bolt 的行为，你可以在提交之前创建拓扑时这样做，你可以通过在 bolt 或 spout 上设置实例变量来实现，从而驱动你想要的可配置行为。

【讨论】：

【解决方案3】：

我也遇到了同样的问题。我通过在集群中配置 NFS 解决了这个问题，并将该配置文件放在共享位置，以便它可用于所有集群机器。在 linux 系统中配置 NFS 非常容易@987654321 @。

【讨论】：

【解决方案4】：

您可以指定与拓扑一起提交的配置（通常通过 yaml 文件）。我们如何在自己的项目中自己管理这个是我们有单独的配置文件用于开发和一个用于生产，在其中我们存储我们的服务器、redis 和数据库 IP 和端口等。然后当我们运行我们的命令来构建 jar 并提交风暴拓扑包括正确的配置文件，具体取决于您的部署环境。 Bolt 和 spout 只需从 StormConf 映射中读取它们所需的配置，该映射在 Bolt 的 prepare() 方法中传递给它们。

来自http://storm.apache.org/documentation/Configuration.html：

Storm 代码库中的 defaults.yaml 中定义了每个配置的默认值。您可以通过在 Nimbus 和主管的类路径中定义一个storm.yaml 来覆盖这些配置。最后，您可以在使用 StormSubmitter 时定义与拓扑一起提交的特定于拓扑的配置。但是，特定于拓扑的配置只能覆盖以“拓扑”为前缀的配置。

Storm 0.7.0 及更高版本允许您基于每个螺栓/每个喷嘴覆盖配置。

您还将在 http://nathanmarz.github.io/storm/doc/backtype/storm/StormSubmitter.html 上看到 submitJar 和 submitTopology 传递了一个名为 conf 的映射。

希望这能让你开始。

【讨论】：

【解决方案5】：

我遇到了和你一样的问题，这是我的棘手解决方案：

使用一个简单的java文件作为配置文件，比如topo_config.java，它看起来像：

package com.xxx
public class topo_config {
    public static String zk_host = "192.168.10.60:2181";
    public static String kafka_topic = "my_log_topic";
    public static int worker_num = 2;
    public static int log_spout_num = 4;
    // ...
}

这个文件放在我的configure文件夹中，然后写一个脚本，比如compile.sh，它会将它复制到正确的包中并进行编译，如下所示：

cp config/topo_config.java src/main/java/com/xxx/
mvn package

直接实现配置：

Config conf = new Config();
conf.setNumWorkers(topo_config.worker_num);

【讨论】：

【解决方案6】：

我通过在代码中提供配置解决了这个问题：

config.put(Config.TOPOLOGY_WORKER_CHILDOPTS, SOME_OPTS);

我尝试提供特定于拓扑的storm.yaml，但它不起作用。如果您可以使用storm.yaml，请纠正我。

更新：
对于任何想知道 SOME_OPTS 是什么的人，这来自 Storm 邮件列表上的 Satish Duggana：

Config.TOPOLOGY_WORKER_CHILDOPTS：可以覆盖的选项 WORKER_CHILDOPTS 用于拓扑。您可以配置任何 java 选项比如内存、gc等

在你的情况下，它可以是

config.put(Config.TOPOLOGY_WORKER_CHILDOPTS, "-Xmx1g");

【讨论】：

storm.yaml 文件不适用于拓扑特定的配置。

【解决方案7】：

我们已经看到了同样的问题，并通过添加以下每个拓扑来解决它

config.put(Config.TOPOLOGY_WORKER_CHILDOPTS, "-Xmx4096m -XX:+UseConcMarkSweepGC -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:NewSize=128m -XX:CMSInitiatingOccupancyFraction=70 -XX:-CMSConcurrentMTEnabled -Djava.net.preferIPv4Stack=true");

还使用 Nimbus UI 进行了验证，如下所示。

topology.worker.childopts   -Xmx4096m -XX:+UseConcMarkSweepGC -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:NewSize=128m -XX:CMSInitiatingOccupancyFraction=70 -XX:-CMSConcurrentMTEnabled -Djava.net.preferIPv4Stack=true

【讨论】：