打印自Apache Storm Bolt

Question

public class ReportBolt extends BaseRichBolt {

  private static final long serialVersionUID = 6102304822420418016L;

  private Map<String, Long> counts;
  private OutputCollector collector;

  @Override @SuppressWarnings("rawtypes")
  public void prepare(Map stormConf, TopologyContext context, OutputCollector outCollector) {
    collector = outCollector;
    counts = new HashMap<String, Long>();
  }

  @Override

  public void declareOutputFields(OutputFieldsDeclarer declarer) {
    // terminal bolt = does not emit anything
  }

  @Override
  public void execute(Tuple tuple) {   
    System.out.println("HELLO " + tuple);
  }

  @Override
  public void cleanup() {
    System.out.println("HELLO FINAL");
  }
}

简而言之，为什么我的println没有显示在工作日志中？

编辑

public class AckedTopology {

  private static final String SPOUT_ID = "monitoring_test_spout";
  private static final String REPORT_BOLT_ID = "acking-report-bolt";
  private static final String TOPOLOGY_NAME = "monitoring-topology";

  public static void main(String[] args) throws Exception {
    int numSpoutExecutors = 1;

    KafkaSpout kspout = buildKafkaSpout();
    ReportBolt reportBolt = new ReportBolt();

    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout(SPOUT_ID, kspout, numSpoutExecutors);
    builder.setBolt(REPORT_BOLT_ID, reportBolt);

    Config cfg = new Config();
    StormSubmitter.submitTopology(TOPOLOGY_NAME, cfg, builder.createTopology());
  }

  private static KafkaSpout buildKafkaSpout() {
    String zkHostPort = "URL";
    String topic = "TOPIC";

    String zkRoot = "/brokers";
    String zkSpoutId = "monitoring_test_spout_id";
    ZkHosts zkHosts = new ZkHosts(zkHostPort);

    SpoutConfig spoutCfg = new SpoutConfig(zkHosts, topic, zkRoot, zkSpoutId);
    KafkaSpout kafkaSpout = new KafkaSpout(spoutCfg);
    return kafkaSpout;
  }
}

Answer 1

你的螺栓没有用链子拴在壶嘴上。你需要使用Storm的分组才能做到这一点..用这样的东西

    builder.setBolt(REPORT_BOLT_ID, reportBolt).shuffleGrouping(SPOUT_ID);

setbolt通常返回InputDeclarer对象。在您的例子中，通过指定shufflegrouping(SPOUT_ID)，您告诉storm您有兴趣使用id为report_bolt_id的组件发出的所有元组。

阅读更多关于流分组的内容，并根据您的需要选择一个。