Field Grouping for a Kafka Spout

问题:

Can field grouping be done on tuples emitted by a kafka spout? If yes, then how does Storm gets to know the fields in a Kafka record?

回答1:

Kafka Spout declared its output fields like any other component. My explanation is based on current implementation of KafkaSpout.

In KafkaSpout.java class we see declareOutputFields method that call getOutputFields() method of KafkaConfig Scheme.

@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(_spoutConfig.scheme.getOutputFields());
}

By default, KafkaConfig uses RawMultiScheme that implements this method in this way.

  @Override
  public Fields getOutputFields() {
    return new Fields("bytes");
  }

So what does it mean?, if you declared bolt which reads tuples from KafkaSpout with fieldGrouping you know that every tuple that contains equals field "bytes" is going to be executed by the same task. If you want to emit any field, you should implement new scheme for your needs.

回答2:

Field grouping (and grouping in general) in Storm is for bolts, not for spouts. This is done via InputDeclarer class.
When you call setBolt() on TopologyBuilder, InputDeclarer is returned.