我如何实现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")); 

享受