Kafka Streams 在 HDFS 上查找数据

Posted 2023-03-23

【中文标题】Kafka Streams 在 HDFS 上查找数据

【英文标题】：Kafka Streams with lookup data on HDFS

【发布时间】：2017-01-10 17:23:15

【问题描述】：

我正在使用 Kafka Streams (v0.10.0.1) 编写应用程序，并希望使用查找数据来丰富我正在处理的记录。该数据（带时间戳的文件）每天（或每天 2-3 次）写入 HDFS 目录。

如何在Kafka Streams 应用程序中加载它并加入实际的KStream？ 当新文件到达那里时，从 HDFS 重新读取数据的最佳做法是什么？

或者切换到Kafka Connect 并将RDBMS 表内容写入Kafka 主题会更好吗？所有Kafka Streams 应用程序实例都可以使用该主题？

更新： 正如建议的那样 Kafka Connect 将是要走的路。因为查找数据在 RDBMS 中每天更新，所以我正在考虑按计划运行 Kafka Connect one-off job，而不是保持连接始终打开。是的，因为语义和保持连接始终打开并确保它不会被中断的开销......等等。对我来说，在这种情况下进行预定提取看起来更安全。

查找数据不大，记录可能被删除/添加/修改。我也不知道如何始终将完整转储到 Kafka 主题并截断以前的记录。启用日志压缩并为已删除的键发送空值可能不起作用，因为我不知道源系统中已删除什么。此外，我无法控制压缩发生时的情况。

【问题讨论】：

我对 Kafka Connect 的细节不熟悉，但是使用它来将数据放入主题中是推荐使用 Kafka Streams 处理它的方式。但是，如果 Connect 连续更新而不是每天更新，对您的情况会不利吗？作为替代方案，您始终可以使用允许执行任意代码的处理器 API。但是，它使用起来要花很多力气。

【参考方案1】：

推荐的方法确实是将查找数据也摄取到 Kafka 中 - 例如通过 Kafka Connect - 正如您在上面自己建议的那样。

但在这种情况下，我如何安排 Connect 作业每天运行，而不是连续从源表中获取，这在我的情况下是不必要的？

也许您可以更新您不希望连续运行 Kafka Connect 作业的问题？您是否关心资源消耗（数据库上的负载），如果不是“每日更新”，您是否关心处理的语义，或者......？

更新： 正如所建议的那样，Kafka Connect 将是要走的路。因为查找数据每天都会在 RDBMS 中更新，所以我考虑将 Kafka Connect 作为计划的一次性作业运行，而不是保持连接始终打开。是的，因为语义和保持连接始终打开并确保它不会被中断的开销......等等。对我来说，在这种情况下进行预定提取看起来更安全。

Kafka Connect 是安全的，并且 JDBC 连接器的构建正是为了以稳健、容错和高性能的方式将数据库表馈送到 Kafka（已经有许多生产部署）。所以我建议不要仅仅因为“看起来更安全”而回退到“批量更新”模式；就个人而言，我认为触发每日摄取在操作上不如让它运行以进行连续（和实时！）摄取更方便，而且它还会为您的实际用例带来一些不利因素（请参阅下一段）。

但是，当然，您的里程可能会有所不同 - 因此，如果您打算每天只更新一次，那就去吧。但是您失去了 a) 在扩充发生的时间点使用最新数据库数据扩充传入记录的能力，并且相反，b) 您实际上可能会使用陈旧/旧数据来扩充传入记录，直到下一天更新已完成，这很可能会导致您向下游发送的数据不正确/可供其他应用程序使用以供使用。例如，如果客户更新了她的送货地址（在数据库中），但您每天只将此信息提供给您的流处理应用程序（可能还有许多其他应用程序），那么订单处理应用程序会将包裹运送到错误的直到下一次每日摄取完成为止。

查找数据不大，可能会删除/添加/修改记录。我也不知道如何始终将完整转储到 Kafka 主题并截断以前的记录。启用日志压缩并为已删除的键发送空值可能不起作用，因为我不知道源系统中已删除的内容。

Kafka Connect 的 JDBC 连接器已经为您自动处理：1. 它确保数据库插入/更新/删除正确反映在 Kafka 主题中，以及 2. Kafka 的日志压缩确保目标主题不会增长出界。也许您可能想阅读文档中的 JDBC 连接器以了解您可以免费获得哪些功能：http://docs.confluent.io/current/connect/connect-jdbc/docs/？