0

さまざまな Kafka トピックから複数の OpaqueTridentKafkaSpout を読み取っています。これらすべてのストリームからのデータが同じ一連の関数を通過するようにします。それを達成するための最良の方法は何ですか。別のストリームを作成し、各タプルを同じ関数セットに再度渡す必要がありますか? 以下のように？

BrokerHosts zk = new ZkHosts(getZooKeeperHosts());
TridentKafkaConfig spoutConf = new TridentKafkaConfig(zk, "Test");
spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
TridentKafkaConfig spoutConf1 = new TridentKafkaConfig(zk, "Test1");
spoutConf1.scheme = new SchemeAsMultiScheme(new StringScheme());
OpaqueTridentKafkaSpout kafkaSpout1 = new OpaqueTridentKafkaSpout(spoutConf1);

topology.newStream("event", kafkaSpout).each(new Fields("document"), new ExtractDocumentInfo(), new Fields("id", "index", "type"));
topology.newStream("event1", kafkaSpout1).each(new Fields("document"), new ExtractDocumentInfo(), new Fields("id", "index", "type"));
4

1 に答える 1

0

ストリームを一緒にマージすることはできますが、失敗すると両方のスパウトがバッチを再生することになります。

于 2015-08-09T15:45:52.660 に答える