我正在尝试编写一个 Flink 应用程序，它从 Kafka 读取事件，从 MySQL 丰富这些事件并将这些数据写入 HBase。我正在RichFlatMapFunction 中进行 MySQL 扩充，现在我正在尝试找出如何最好地写入 HBase。我想批量写入 HBase，所以我目前正在考虑使用WindowAll，然后是身份apply（只是为了回到DataStream），然后写一个HBaseSink，它需要一个列表记录并批量处理Puts 到HBase。
这是正确的做事方式吗？使用WindowAll 和apply 只是为了进行基于时间的缓冲感觉很奇怪。
【问题讨论】：
标签： apache-flink flink-streaming
从我在 Jira 上看到的 FLINK-2055 来看，我认为您目前最好的选择是使用 Flink Streaming 的 support for Async I/O，并在该自定义函数中处理缓冲。看起来棘手的部分来自（a）正确处理检查点/重试（原子放置）和（b）避免 HBase 区域服务器过载。但是，如果您不担心完全一次支持，并且可以根据您的 HBase 设置调整设置，那么这应该非常简单。
【讨论】：
与 HBase 打交道时最繁重的操作实际上是打开与它的连接（在文档中也有说明：https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/Connection.html）。
我的建议是实现另一个RichFlatMapFunction，这样您就可以在每个任务中只打开一个连接一次，然后每次需要持久化事件时对HBase 执行Put。
到目前为止，与您当前方法的不同之处在于：
WindowAll
HBaseSink 中打开到 HBase 的新连接
关于从批处理或从流的角度执行 Put 到 HBase 的事实，实际上并没有改变操作的复杂性（您也可以从这个问题的第一个答案中看到：Is HBase batch put put(List<Put>) faster than put(Put)? What is the capacity of a Put object? ) 到目前为止，您仍将执行 N 操作以持久化 N 事件。
【讨论】：
RichSinkFunction 对吗？这也让我可以覆盖 open 方法，并且我只能在其中打开一次连接。
RichFlatMapFunction 相同但以完全异步的方式，那么你可以使用@kkrugler 建议的RichAsyncFuntion您可以查看ci.apache.org/projects/flink/flink-docs-stable/dev/stream/… 的文档，然后您可以在open() 方法的覆盖中打开与HBase 的连接。