我刚开始接触Apache Storm。我阅读了教程并查看了examples。我的问题是所有的例子都使用非常简单的元组(通常是一个带有字符串的字段)。这些元组是内联创建的(使用new Values(...))。在我的情况下,我有许多字段(5..100)的元组。所以我的问题是如何为每个字段实现具有名称和类型(所有基本类型)的元组?
是否有任何示例?(我认为直接实现“Tuple”不是一个好主意)
谢谢
是否有任何示例?(我认为直接实现“Tuple”不是一个好主意)
谢谢
将所有字段作为值创建元组的替代方法是只需创建一个 bean 并将其传递到元组中。
给定以下类:
public class DataBean implements Serializable {
private static final long serialVersionUID = 1L;
// add more properties as necessary
int id;
String word;
public DataBean(int id, String word) {
setId(id);
setWord(word);
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getWord() {
return word;
}
public void setWord(String word) {
this.word = word;
}
}
在一个bolt中创建并发射DataBean:
collector.emit(new Values(bean));
在目标 bolt 中获取 DataBean:
@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
try {
DataBean bean = (DataBean)tuple.getValue(0);
// do your bolt processing with the bean
} catch (Exception e) {
LOG.error("WordCountBolt error", e);
collector.reportError(e);
}
}
不要忘记在设置拓扑时使你的bean可序列化并进行注册:
Config stormConfig = new Config();
stormConfig.registerSerialization(DataBean.class);
// more stuff
StormSubmitter.submitTopology("MyTopologyName", stormConfig, builder.createTopology());
免责声明:对于shuffle分组,豆子将正常工作。如果您需要进行fieldsGrouping
,仍应使用原始值。例如,在单词计数场景中,您需要按单词分组,因此可以发出以下内容:
collector.emit(new Values(word, bean));
Values
类型的对象列表中的固定索引。这种方法避免了一个普通Bean所面临的“字段分组”问题。
以单词计数为例,代码示例如下:
public class WordCountTuple extends Values {
private final static long serialVersionUID = -4386109322233754497L;
// attribute indexes
/** The index of the word attribute. */
public final static int WRD_IDX = 0;
/** The index of the count attribute. */
public final static int CNT_IDX = 1;
// attribute names
/** The name of the word attribute. */
public final static String WRD_ATT = "word";
/** The name of the count attribute. */
public final static String CNT_ATT = "count";
// required for serialization
public WordCountTuple() {}
public WordCountTuple(String word, int count) {
super.add(WRD_IDX, word);
super.add(CNT_IDX, count);
}
public String getWord() {
return (String)super.get(WRD_IDX);
}
public void setWort(String word) {
super.set(WRD_IDX, word);
}
public int getCount() {
return (Integer)super.get(CNT_IDX);
}
public void setCount(int count) {
super.set(CNT_IDX, count);
}
public static Fields getSchema() {
return new Fields(WRD_ATT, CNT_ATT);
}
}
final static
变量。此外,方法getSchema()
返回实现的模式,用于在Spout/Bolt方法.declareOutputFields(...)
中声明输出流。public MyOutBolt implements IRichBolt {
@Override
public void execute(Tuple tuple) {
// some more processing
String word = ...
int cnt = ...
collector.emit(new WordCountTuple(word, cnt));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(WordCountTuple.getSchema());
}
// other methods omitted
}
对于输入元组,我建议采用以下模式:
public MyInBolt implements IRichBolt {
// use a single instance for avoid GC trashing
private final WordCountTuple input = new WordCountTuple();
@Override
public void execute(Tuple tuple) {
this.input.clear();
this.input.addAll(tuple.getValues());
String word = input.getWord();
int count = input.getCount();
// do further processing
}
// other methods omitted
}
MyOutBolt
和MyInBolt
可以如下连接:
TopologyBuilder b = ...
b.setBolt("out", new MyOutBolt());
b.setBolt("in", new MyInBolt()).fieldsGrouping("out", WordCountTuple.WRD_ATT);
WordCountTuple
允许分别访问每个属性。null
初始化可能是一个有效的改进。但对于设置器来说,这无关紧要,因为它们使用索引访问。当然,必须小心使用常规的数组列表方法(Values
继承自那个类)。否则,你可能会把一切搞砸! - Matthias J. Sax