Как изменить ESCrawlTopology так он работает на местном вместо удаленного? исключение «NoNodeAvailableException»

Вопрос задан: 1 год назад Последняя активность: 1 год назад
up 0 down

Я по существу хочу повторить эту команду:

storm jar target/crawlIndexer-1.0-SNAPSHOT.jar 
org.apache.storm.flux.Flux es-crawler.flux --local --sleep 30000

но сделать его исполняемым класс (по аналогии с ESCrawlToplogy). Но сделать это местным

Я попытался это до сих пор:

public class ESCrawlTopology extends ConfigurableTopology {

public static void main(String[] args) throws Exception {
    ConfigurableTopology.start(new ESCrawlTopology(), 
  new String[]{"-conf", "crawler-conf.yaml","-local"}); //Added local flag
}

@Override
protected int run(String[] args) {
    TopologyBuilder builder = new TopologyBuilder();

    int numWorkers = ConfUtils.getInt(getConf(), "topology.workers", 1);

    // set to the real number of shards ONLY if es.status.routing is set to
    // true in the configuration
    int numShards = 1;

    builder.setSpout("spout", new CollapsingSpout(), numShards);

    builder.setBolt("status_metrics", new StatusMetricsBolt())
            .shuffleGrouping("spout");

    builder.setBolt("partitioner", new URLPartitionerBolt(), numWorkers)
            .shuffleGrouping("spout");

    builder.setBolt("fetch", new FetcherBolt(), numWorkers).fieldsGrouping(
            "partitioner", new Fields("key"));

    builder.setBolt("sitemap", new SiteMapParserBolt(), numWorkers)
            .localOrShuffleGrouping("fetch");

    builder.setBolt("parse", new JSoupParserBolt(), numWorkers)
            .localOrShuffleGrouping("sitemap");

    builder.setBolt("indexer", new IndexerBolt(), numWorkers)
            .localOrShuffleGrouping("parse");

    Fields furl = new Fields("url");

    builder.setBolt("status", new StatusUpdaterBolt(), numWorkers)
            .fieldsGrouping("fetch", Constants.StatusStreamName, furl)
            .fieldsGrouping("sitemap", Constants.StatusStreamName, furl)
            .fieldsGrouping("parse", Constants.StatusStreamName, furl)
            .fieldsGrouping("indexer", Constants.StatusStreamName, furl);

    builder.setBolt("deleter", new DeletionBolt(), numWorkers)
            .localOrShuffleGrouping("status",
                    Constants.DELETION_STREAM_NAME);

    conf.registerMetricsConsumer(MetricsConsumer.class);
    conf.registerMetricsConsumer(LoggingMetricsConsumer.class);


    return submit("crawl", conf, builder);
}

Основное изменение, что я сделал это добавление «-local» флаг в качестве аргумента основного метода.

Выше, кажется, чтобы загрузить шторм локально успешно, однако я получаю сообщение об ошибке с ElasticSearch.

org.elasticsearch.client.transport.NoNodeAvailableException: None of the configured nodes are available: []
    at org.elasticsearch.client.transport.TransportClientNodesService.ensureNodesAreAvailable(TransportClientNodesService.java:344) ~[elasticsearch-5.3.0.jar:5.3.0]
    at org.elasticsearch.client.transport.TransportClientNodesService.execute(TransportClientNodesService.java:242) ~[elasticsearch-5.3.0.jar:5.3.0]
    at org.elasticsearch.client.transport.TransportProxyClient.execute(TransportProxyClient.java:59) ~[elasticsearch-5.3.0.jar:5.3.0]
    at org.elasticsearch.client.transport.TransportClient.doExecute(TransportClient.java:366) ~[elasticsearch-5.3.0.jar:5.3.0]
    at org.elasticsearch.client.support.AbstractClient.execute(AbstractClient.java:404) ~[elasticsearch-5.3.0.jar:5.3.0]
    at org.elasticsearch.action.ActionRequestBuilder.execute(ActionRequestBuilder.java:80) ~[elasticsearch-5.3.0.jar:5.3.0]
    at com.digitalpebble.stormcrawler.elasticsearch.persistence.CollapsingSpout.populateBuffer(CollapsingSpout.java:140) ~[storm-crawler-elasticsearch-1.5.jar:?]
    at com.digitalpebble.stormcrawler.elasticsearch.persistence.AbstractSpout.nextTuple(AbstractSpout.java:322) ~[storm-crawler-elasticsearch-1.5.jar:?]
    at org.apache.storm.daemon.executor$fn__4976$fn__4991$fn__5022.invoke(executor.clj:644) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.util$async_loop$fn__557.invoke(util.clj:484) [storm-core-1.1.0.jar:1.1.0]
    at clojure.lang.AFn.run(AFn.java:22) [clojure-1.7.0.jar:?]
    at java.lang.Thread.run(Thread.java:748) [?:1.8.0_151]

Любые идеи? Благодарность

1 ответ

Возможно, для Вашего проекта будут необходимы бесплатные векторные карты. На нашем сайте представлены карты для всех стран.

Реклама

up 0 down accepted

Решаемые.

Просто пришлось добавить ElasticSearch конф файл в аргументах.

Так:

public static void main(String[] args) throws Exception {
    ConfigurableTopology.start(new ESCrawlTopology(), 
  new String[]{"-conf", "es-conf.yaml","-local"});
}