Настройка топологии шторма
Как вы предоставляете пользовательскую конфигурацию для топологии шторма? Например, если у меня есть построенная топология, которая подключается к кластеру MySQL, и я хочу иметь возможность изменять, к каким серверам мне нужно подключаться, без перекомпиляции, как мне это сделать? Я предпочел бы использовать файл конфигурации, но меня беспокоит то, что сам файл не развертывается в кластере, поэтому он не будет запущен (если мое понимание того, как работает кластер, не является ошибочным). До сих пор я видел единственный способ передачи параметров конфигурации в топологию шторма во время выполнения - через параметр командной строки, но это беспорядочно, когда вы получаете достаточное количество параметров.
Одна мысль действительно имела место - использовать сценарий оболочки, чтобы прочитать файл в переменную и передать содержимое этой переменной в виде строки в топологию, но я хотел бы, чтобы это было немного чище.
кто-нибудь еще сталкивался с этим? Если да, то как ты решил это?
РЕДАКТИРОВАТЬ:
Похоже, необходимо предоставить больше разъяснений. Мой сценарий состоит в том, что у меня есть топология, которую я хочу иметь возможность развертывать в разных средах без необходимости перекомпилировать ее. Обычно я создаю файл конфигурации, который содержит такие вещи, как параметры соединения с базой данных, и передает их. Я хотел бы знать, как сделать что-то подобное в Storm.
7 ответов
Вы можете указать конфигурацию (обычно через файл yaml), которую вы отправляете вместе с топологией. Как мы сами справляемся с этим в нашем собственном проекте, так это то, что у нас есть отдельные файлы конфигурации для разработки и один для производства, и внутри него мы храним наш сервер, IP-адреса redis и db, порты и т. Д. Затем, когда мы запускаем нашу команду, чтобы собрать jar и отправить топология штурма включает в себя правильный файл конфигурации в зависимости от среды развертывания. Болты и носики просто читают конфигурацию, которая им требуется, из карты stormConf, которая передается им в методе prepare() вашего болта.
С http://storm.apache.org/documentation/Configuration.html:
Каждая конфигурация имеет значение по умолчанию, определенное в defaults.yaml в кодовой базе Storm. Вы можете переопределить эти конфигурации, определив storm.yaml в пути к классам Nimbus и супервизоров. Наконец, вы можете определить специфическую для топологии конфигурацию, которую вы отправляете вместе с топологией при использовании StormSubmitter. Однако конфигурация, относящаяся к топологии, может переопределять только конфиги с префиксом "TOPOLOGY".
Storm 0.7.0 и более поздние версии позволяют переопределять конфигурацию для каждого болта / носика.
На http://nathanmarz.github.io/storm/doc/backtype/storm/StormSubmitter.html вы также увидите, что submitJar и submitTopology передаются карте с именем conf.
Надеюсь, это поможет вам начать.
Я решил эту проблему, просто предоставив конфиг в коде:
config.put(Config.TOPOLOGY_WORKER_CHILDOPTS, SOME_OPTS);
Я пытался указать топологию storm.yaml но это не работает Поправьте меня, если вы заставите его работать на storm.yaml.
Обновить:
Для тех, кто хочет знать, что такое SOME_OPTS, это от Сатиша Дугганы из списка рассылки Storm:
Config.TOPOLOGY_WORKER_CHILDOPTS: параметры, которые могут переопределять WORKER_CHILDOPTS для топологии. Вы можете настроить любые параметры Java, такие как память, GC и т. Д.
В вашем случае это может быть
config.put(Config.TOPOLOGY_WORKER_CHILDOPTS, "-Xmx1g");
На самом деле, вам лучше всего сохранить конфигурацию в хранилище изменяемых значений ключей (s3, redis и т. Д.), А затем использовать его для настройки соединения с базой данных, которое вы затем используете (я предполагаю, что вы уже планируете ограничить то, как часто вы общаетесь с базой данных, так что затраты на эту конфигурацию не имеют большого значения). Такая конструкция позволяет изменять соединение с базой данных на лету без необходимости даже повторного развертывания топологии.
Мы видели ту же проблему и решили ее, добавив приведенную ниже топологию
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, как показано ниже.
topology.worker.childopts -Xmx4096m -XX:+UseConcMarkSweepGC -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:NewSize=128m -XX:CMSInitiatingOccupancyFraction=70 -XX:-CMSConcurrentMTEnabled -Djava.net.preferIPv4Stack=true
Я сталкиваюсь с той же проблемой, что и вы, и вот мое хитрое решение:
Используйте простой файл 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;
// ...
}
Этот файл помещается в мою папку конфигурации, а затем написать сценарий, скажем, compile.sh который скопирует его в нужный пакет и выполнит компиляцию, выглядит так:
cp config/topo_config.java src/main/java/com/xxx/
mvn package
Конфигурация достигается напрямую:
Config conf = new Config();
conf.setNumWorkers(topo_config.worker_num);
Идея состоит в том, что когда вы строите свою топологию, вы создаете экземпляры своих патрубков и болтов (среди прочего), и эти экземпляры сериализуются и распределяются по нужным местам в кластере. Если вы хотите настроить поведение носика или болта, вы делаете это при создании топологии перед отправкой, и вы делаете это, устанавливая переменные экземпляра на болте или носике, которые, в свою очередь, управляют желаемым настраиваемым поведением.
Я также столкнулся с той же проблемой. Я решил ее, настроив NFS в своем кластере, и поместил этот файл конфигурации в общее местоположение, чтобы он был доступен для всех компьютеров кластера. Настроить NFS в системе linux очень просто.