使用Apache Flink流处理缓冲转换后的消息（例如，1000计数）

Question

我正在使用Apache Flink进行流处理。

订阅源消息（例如：Kafka、AWS Kinesis数据流）后，然后使用Flink操作符对流数据应用转换、聚合等，我想缓冲最终消息（例如：1000条），并在单个请求中将每个批发布到外部REST API。

如何在Apache Flink中实现缓冲机制（将每1000条记录创建为批处理）？

Flink Pipiline：流媒体源--

感谢您的帮助！

Answer 1

我将创建一个具有状态的接收器，该接收器将保留传入的消息。当计数足够高（1000）时，接收器发送批。状态可以在内存中（例如，一个实例变量保存消息的ArrayList），但您应该使用检查点，以便在出现某种故障时可以恢复该状态。

当您的接收器具有检查点状态时，它需要实现Checkpoint edFunction（org.apache.flink.streaming.api.checkpoint），这意味着您需要向接收器添加两个方法：

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {

    checkpointedState.clear();

    // HttpSinkStateItem is a user-written class 
    // that just holds a collection of messages (Strings, in this case)
    //
    // Buffer is declared as ArrayList<String>

    checkpointedState.add(new HttpSinkStateItem(buffer));

}

@Override
public void initializeState(FunctionInitializationContext context) throws Exception {

    // Mix and match different kinds of states as needed:
    //   - Use context.getOperatorStateStore() to get basic (non-keyed) operator state
    //        - types are list and union        
    //   - Use context.getKeyedStateStore() to get state for the current key (only for processing keyed streams)
    //        - types are value, list, reducing, aggregating and map
    //   - Distinguish between state data using state name (e.g. "HttpSink-State")      

    ListStateDescriptor<HttpSinkStateItem> descriptor =
        new ListStateDescriptor<>(
            "HttpSink-State",
            HttpSinkStateItem.class);

    checkpointedState = context.getOperatorStateStore().getListState(descriptor);

    if (context.isRestored()) {

        for (HttpSinkStateItem item: checkpointedState.get()) {
            buffer = new ArrayList<>(item.getPending());  
        }

    }       

}

如果计数未达到阈值，还可以使用接收器中的计时器（如果输入流已设置关键帧/分区）定期发送。