如何使用非键控状态与Kafka消费者在Flink？

Question

我试图在KafkaConsumer对象中实现（刚刚开始使用Java和Flink）一个非键控状态，因为在这个阶段没有调用keyBy（）。该对象是前端，也是处理来自Kafka的消息的第一个模块。SourceOutput是一个表示消息的proto文件。

public class KafkaSourceFunction extends ProcessFunction<byte[], SourceOutput> implements Serializable
{
    @Override
    public void processElement(byte[] bytes, ProcessFunction<byte[], SourceOutput>.Context 
         context, Collector<SourceOutput> collector) throws Exception
    {
          // Here, I want to call to sorting method
          collector.collect(output);
    }
}

class SessionInfo
{
    public PriorityQueue<SourceOutput>  orderedMessages = null;

    public void putMessage(SourceOutput Msg)
    {
        if(orderedMessages == null)
            orderedMessages = new PriorityQueue<SourceOutput>(new SequenceComparator());

        orderedMessages.add(Msg);
    }
}

public class KafkaSourceState  implements Serializable
{
    public TreeMap<String, SessionInfo> Sessions = new TreeMap<>();
}

public class KafkaSourceSort implements SinkFunction<KafkaSourceSort>,
        CheckpointedFunction
{
    private transient ListState<KafkaSourceState> checkpointedState;
    private KafkaSourceState state;

    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception
    {
        checkpointedState.clear();
        checkpointedState.add(state);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception
    {
        ListStateDescriptor<KafkaSourceState> descriptor =
                new ListStateDescriptor<KafkaSourceState>(
                        "KafkaSourceState",
                        TypeInformation.of(new TypeHint<KafkaSourceState>() {}));

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

        if (context.isRestored())
        {
            state = (KafkaSourceState) checkpointedState.get();
        }
    }

    @Override
    public void invoke(KafkaSourceState value, SinkFunction.Context contex) throws Exception
    {
        state = value;

       // ...
    }
}

我看到我需要实现一个调用消息，它可能会从processElement（）调用，但是invoke（）的签名不包含收集器，我不知道如何这样做，甚至到目前为止我做得还好。

请帮忙，不胜感激。谢了。

Answer 1

sinkfunction是作业图DAG中的终端节点。它的接口中没有收集器，因为它不能向下游发出任何东西。它被期望连接到外部服务或数据存储并在那里发送数据。

如果你更多地分享你正在努力完成的事情，也许我们可以提供更多的帮助。也许有一个更简单的方法来解决这个问题。