我正在研究一个将流写入关系数据库的flink项目。
在当前的解决方案中，我们编写了一个自定义接收函数，它打开事务，执行SQL插入语句和关闭事务。它运行良好，直到数据量增加，我们开始获得连接超时问题。我们尝试了一些连接池配置调整，它没有多大帮助。
我们正在考虑尝试“批量插入”，以减少对数据库的“写入”次数。我们遇到了几个几乎可以做我们想要的类：JDBCOutputFormat，JDBCSinkFunction。使用JDBCOutputFormat，我们可以配置批量大小。
如果记录数量不超过“批量大小”，我们还希望每1分钟强制进行一次“批量插入”。你通常会如何处理这些问题？我的第一个想法是扩展JDBCOutputFormat以使用计划任务每1分钟强制刷新一次，但是如何完成它并不明显。
我们是否必须一起编写自己的接收器？
更新：
每次Flink检查点时，JDBCSinkFunction都会执行刷新和批处理执行。只要您进行检查点检查，批次将不会超过检查点间隔。
但是，在阅读了this mailing list thread后，我发现JDBCSinkFunction不支持完全一次输出。