Apache Storm 和 Kafka 与 Firebase 的集成答案

【问题标题】：Apache Storm and Kafka integration with FirebaseApache Storm 和 Kafka 与 Firebase 的集成
【发布时间】：2019-07-18 07:23:37
【问题描述】：

    import backtype.storm.Config;`enter code here`
    import backtype.storm.`StormSubmitter`;`enter code here
    import backtype.storm.topology.TopologyBuilder;`enter code here`
    import consumer.bolt.FirebaseBolt;`enter code here`
    import consumer.bolt.WordCountBolt;
    import consumer.bolt.WordCountDumpBolt;
    import nl.minvenj.nfi.storm.kafka.KafkaSpout;

    /**
     * User: tonymeng
     * Date: 3/31/14
     */
    public class FirebaseTopologyCluster {

      public static void main(String[]args) throws Exception {
        if (args == null || args.length != 3) {
          throw new IllegalArgumentException("localhost:2181 ,testTopic, https://chem-9b445.firebaseio.com/");
        }
        String zkConnect = args[0];
        String topic = args[1];
        String firebaseNamespace = args[2];

        Config config = new Config();
        config.setNumWorkers(1);

        config.put("kafka.spout.topic", topic);
        config.put("kafka.spout.consumer.group", "test-consumer-group");
        config.put("kafka.zookeeper.connect", zkConnect);
        config.put("kafka.consumer.timeout.ms", 4000);

        KafkaSpout spout = new KafkaSpout();
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("kafkaspout", spout);
        builder.setBolt("countbolt", new WordCountBolt()).shuffleGrouping("kafkaspout");
        builder.setBolt("countfilebolt", new WordCountDumpBolt("/tmp/stats")).shuffleGrouping("countbolt");
        // using '`' as a delimiter
        builder.setBolt("firebasebolt", new FirebaseBolt(firebaseNamespace, "`")).shuffleGrouping("countbolt");

        StormSubmitter.submitTopology("statstopology", config, builder.createTopology());
      }
    }

这是我的代码。在此运行后向我显示此类消息。我在上面写的。我该如何解决这个问题：

线程“main”中的异常 java.lang.IllegalArgumentException: localhost:2181 ,testTopic, chem-9b445

【问题讨论】：

看起来您在尝试运行它时需要提供三个参数（zookeeper、topic 和 firebase 命名空间）。顺便说一句，我认为这不是您的代码。

标签： java firebase-realtime-database apache-kafka apache-storm

【解决方案1】：

看起来你正在抛出那个异常。

if (args == null || args.length != 3) {
    throw new IllegalArgumentException("localhost:2181 ,testTopic, https://chem-9b445.firebaseio.com/");
}

在运行时，您应该传递 3 个参数以正确运行此代码。

这些参数是；

String zkConnect = args[0];
String topic = args[1];
String firebaseNamespace = args[2];

【讨论】：