在storm中使用trident的tick元组

Question

谢谢

Answer 1

实际上，微批量是三叉戟的一个内置功能。你不需要任何刻度元组。当您的代码中包含以下内容时：

topology
    .newStream("myStream", spout)
    .partitionPersist(
        ElasticSearchEventState.getFactoryFor(connectionProvider),
        new Fields("field1", "field2"),
        new ElasticSearchEventUpdater()
    )

（我在这里使用自定义的ElasticSearch状态/更新程序，您可以使用其他的）

因此，当您有这样的东西时，在hood Trident下，将您的流分组为批处理，并对这些批处理而不是单个元组执行partitionPersist操作。

public class TickSpout implements IBatchSpout {

    public static final String TIMESTAMP_FIELD = "timestamp";
    private final long delay;

    public TickSpout(long delay) {
        this.delay = delay;
    }

    @Override
    public void open(Map conf, TopologyContext context) {
    }

    @Override
    public void emitBatch(long batchId, TridentCollector collector) {
        Utils.sleep(delay);
        collector.emit(new Values(System.currentTimeMillis()));
    }

    @Override
    public void ack(long batchId) {
    }

    @Override
    public void close() {
    }

    @Override
    public Map getComponentConfiguration() {
        return null;
    }

    @Override
    public Fields getOutputFields() {
        return new Fields(TIMESTAMP_FIELD);
    }
}