在 DataStream Flink 中从 JDBC 源读取数据的问题

Question

我正在启动一个新的Flink应用程序，允许我的公司执行大量报告。我们有一个现有的遗留系统，我们需要的大部分数据都保存在SQL Server数据库中。在开始使用新部署的Kafka流中的更多数据之前，我们首先需要使用这些数据库中的数据。

我花了很多时间阅读Flink的书和网页，但我有一些简单的问题和假设，我希望你能帮助我进步。

首先，我想使用DataStream API，这样我们既可以使用历史数据，也可以使用实时数据。我不想使用DataSet API，但我也不认为使用SQL/Table API有什么意义，因为我更喜欢用Java类编写函数。我需要维护自己的状态，而数据流键控函数似乎是一条出路。

现在我正试图针对我们的生产数据库编写代码，我需要能够通过SQL查询读入“流”数据-似乎没有JDBC源连接器，因此我认为我必须自己进行JDBC调用，然后可能使用env.fromElements（）创建数据源。显然，这是一个“有界”的数据集，但我怎么才能加载历史数据呢？在未来，我还想包括一个Kafka流，它只有几周的数据，所以我想我有时需要将SQL Server/Snowflake数据库中的数据与Kafka数据流中的实时数据流合并。这方面的最佳实践是什么，因为我没有看到讨论这一点的示例。

在从JDBC数据源检索数据时，我也看到了一些使用StreamingTableEnvironment的例子——我是不是应该用它来查询从JDBC连接到我的数据流函数中的数据？同样，我想用Java写函数，而不是用Flink SQL。如果我只使用DataStream API，那么使用StreamingTableEnvironment查询JDBC数据是否是最佳实践？

Answer 1

可以使用以下方法来读取数据库并创建数据流：

>

您可以使用RichParallelSourceFunction对数据库进行自定义查询并从中获取数据流。可以在RichParallelSourceFunction类的扩展中启动带有JDBC的SQL驱动程序。

使用Table DataStream API-可以通过创建JDBC曲库查询数据库，然后将其转换为流

另一种选择，也许是更昂贵的解决方案 - 您可以使用Flink CDC连接器，它为Apache Flink提供源连接器，使用更改数据捕获（CDC）从不同的数据库引入更改

然后你可以添加Kafka作为源，得到一个数据流。

因此，简单地说，您的管道可以如下所示：您将两个源转换为数据流，您可以使用协进程函数连接这些流，这也将使您能够维护状态并在业务逻辑中使用它。最后，使用Sink函数将最终输出接收到数据库、Kafka甚至AWS S3存储桶。