java - 如何根据消息中的值将元组发送到不同的 bolt

Question

我有一个连接到 Kinesis Stream 的 Storm 集群。消息看起来像这样。

{
    _c: "a"
}

或者应该是

{
    _c: "b"
}

我想将 _c="a"的元组发送到一个 bolt ，将 _c="b"的元组发送到另一个 bolt 。我如何实现这一目标？

这是使用 GSon 将来自 Kinesis 的消息解析为 JSON 对象的 bolt

@Override
public void execute(Tuple tuple) {
  String partitionKey = (String) tuple.getValueByField(SampleKinesisRecordScheme.FIELD_PARTITION_KEY);
  String sequenceNumber = (String) tuple.getValueByField(SampleKinesisRecordScheme.FIELD_SEQUENCE_NUMBER);
  byte[] payload = (byte[]) tuple.getValueByField(SampleKinesisRecordScheme.FIELD_RECORD_DATA);

  ByteBuffer buffer = ByteBuffer.wrap(payload);
  String data = null;
  try {
    data = decoder.decode(buffer).toString();

    HashMap < String, String > map = new Gson().fromJson(data, new TypeToken < HashMap < String, Object >> () {}.getType());

    this.outputCollector.emit(tuple, new Values(map));
    this.outputCollector.ack(tuple);

  } catch (CharacterCodingException e) {
    this.outputCollector.fail(tuple);
  }

}

谢谢

Answer 1

您可以在 bolt 中定义两个流，然后声明两个输出流:

@Override
public void execute(Tuple tuple) {
    // ...
    // Some Code
    // ...
    if (_c =="a") {
    collector.emit("stream1", tuple, new Values(_c));
    } else {
    collector.emit("stream2", tuple, new Values(_c));
    }

}

@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    outputFieldsDeclarer.declareStream("stream1", new Fields("_c"));
    outputFieldsDeclarer.declareStream("stream2", new Fields("_c"));
} }

然后在您的拓扑中，您​​可以使用 ShuffleGrouping 中的选项来传递 Stream_id。

topology.setBolt("FirstBolt",new FirstBolt(),1);    
topology.setBolt("newBolt1", new Custombolt(),1).shuffleGrouping("FirstBolt", "stream1");
topology.setBolt("newBolt2", new Custombolt(),1).shuffleGrouping("FirstBolt", "stream2");

另一种可能性是将它发送到两个 bolt ，然后检查两个 bolt 中的值并执行所需的代码。