Flink：如何使用键/值存储转换 DataStream？

我想使用 Flink 从输入的 Kafka 主题中连续摄取消息，对于每条消息，从消息中获取一个键字段，在键/值存储中进行查找，创建消息的修改版本使用该值，并将结果消息输出到不同的 Kafka 主题。这是非常标准的流处理功能。

键/值存储的实现仍在决定中，可能通过 Redis 或 Aerospike 或 sql 数据库或 Kafka 主题或其他可能的东西来实现。

我将如何在 Flink 中做到这一点？我使用 DataStream API 来读/写 Kafka 主题。我将如何进行键/值查找？我会使用 Table API 吗？ DataStream API？

在其他工具集中，这会被视为流到表的连接吗？ Flink 认为这是一个连接吗？ Flink 文档提到了连接，这个用例将是“常规连接”。我需要使用 Flink sql 吗？我可以使用 Flink API 进行这种类型的流到表连接吗？

我已经浏览了官方示例存储库中的文档和官方代码示例，但我没有看到这个特定的用例。

是否有此类功能的简单代码示例？