我如何實現KafkaSpout和螺栓之間的字段分組

Question

我一直在向Kafka主題發送消息，這些消息在主題中使用JSON，我使用KafkaSpout從Kafka獲取消息並使用shuffle將消息發送給螺栓分組。現在我想要在KafkaSpout和bolt之間實現字段組合。請任何人都可以幫我解決這個問題。如何實現KafkaSpout與螺栓之間的區域分組。

Answer 1

您需要實現backtype.storm.spout.scheme接口，基本上它看起來是這樣的：

public class FooScheme implements Scheme { 

    public Values deserialize(final byte[] _line) { 

    try{ 
      Values values = new Values(); 
      JSONObject msg = (JSONObject) JSONValue.parseWithException(new String(_line)); 
      values.add((String) msg.get("a")); 
      values.add((String) msg.get("b")) 
      values.add(msg) 
     } 
     catch(ParseException e) { 
      //handle the exception 
      return null; 
     } 

    } 

    public Fields getOutputFields() { 
    return new Fields("a", "b", "json");} 
} 

，你與你的嘴像這樣使用：

SpoutConfig spoutConfig = new SpoutConfig(... your config here ...); 
spoutConfig.scheme = new SchemeAsMultiScheme(new FooScheme()); 
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig); 
topology.setSpout("kafka-spout", 1).setNumTasks(1); 

，現在你可以準備好使用「a」或「b」或兩者的字段分組。

FooBolt bolt = new FooBolt(); 
topology.setBolt("foo-bolt", new FooBolt(), 1).setNumtasks(1) 
     .fieldsGrouping("kafka-spout", new Fields("a","b")); 

享受