大数据如何使用Flink连接openGauss数据库（flink-cdc-connector）

什么是flink-cdc-connector

本文会在最后附上代码修改的git patch

首先简单介绍下flink，Apache Flink是由Apache软件基金会开发的开源流处理框架，其核心是用Java和Scala编写的分布式流数据流引擎。Flink以数据并行和管道方式执行任意流数据程序，Flink的流水线运行时系统可以执行批处理和流处理程序。此外，Flink的运行时本身也支持迭代算法的执行。Flink提供高吞吐量、低延迟的流数据引擎以及对事件-时间处理和状态管理的支持。Flink应用程序在发生机器故障时具有容错能力，并且支持exactly-once语义。程序可以用Java、Scala、Python和SQL等语言编写，并自动编译和优化到在集群或云环境中运行的数据流程序。

常见的场景是，flink将从事件流中获取到的数据进行计算并写入外部的存储系统中以待进一步查询分析。涉及到openGauss这类事务数据库的常见应用场景是，flink任务监听事务数据库（比如MySQL，postgreSQL）的变化数据，实时写入到分析型数据库中以便分析查询可以高效利用分析型数据库性能的同时，也获得数据的实时性。

上面的场景中提到了“变化数据”，那么如果应用如何捕获数据库的变化数据呢？这里要引入一个概念叫做change data capture（CDC）。开源项目debezium实现了大量主流数据库的CDC，虽然没有openGauss的对应实现，不过得益于openGauss对postgreSQL优秀的兼容性，我们可以在postgreSQL实现的基础进行少量修改。但是我们有了CDC后，如何把CDC接入flink呢？这就需要用到另一个项目，也是这篇文章将重点介绍的flink-cdc-connectors。

flink-cdc-connectors这个项目包括多个flink source connector，即把事务数据库作为flink source的connector，而且他依赖了debezium并做了些许修改，我们可以直接在flink-cdc-connectors修改兼容openGauss而不再需要修改debezium了。利用flink-cdc-connectors，就可以通过写SQL来让flink监听openGauss的变化数据，并实时写入ElasticSearch中加速分析查询。

看起来一切都已就绪，但是……

测试使用postgres-connector连接openGauss

参考flink-cdc-connectors的postgres测试文档，首先我们尝试直接使用openGauss来替换掉其中的postgres，看看是否会遇到问题。测试环境的基本信息是：

openGauss 5.0

Flink 1.17.0

flink-sql-connector-elasticsearch7 3.0.1-1.17

flink-sql-connector-mysql-cdc 2.4-SNAPSHOT (build from branch release-2.4)

flink-sql-connector-postgres-cdc 2.4-SNAPSHOT (build from branch release-2.4)

想在本地完成测试，需要对测试文档中flink client建表语句进行修改：

CREATE TABLE shipments (

shipment_id INT,

order_id INT,

origin STRING,

destination STRING,

is_arrived BOOLEAN,

PRIMARY KEY (shipment_id) NOT ENFORCED

) WITH (

'connector' = 'postgres-cdc',

'hostname' = 'localhost',

'port' = 'YOUR_PORT',

'username' = 'YOUR_USERNAME',

'password' = 'YOUR_PASSWORD',

'database-name' = 'postgres',

'schema-name' = 'public',

'table-name' = 'shipments',

'slot.name' = 'your_slot_name',

'decoding.plugin.name' = 'pgoutput',

'debezium.publication.autocreate.mode' = 'filtered'

);

主要是增加了slot.name，修改decoding.plugin.name插件为pgoutput，参考Debezium connector for PostgreSQL中debezium的建议，修改publication.autocreate.mode为filtered避免权限问题。

为了使flink可以连接到openGauss，也需要修改openGauss的配置

gs_guc set -D ${PGDATA} -c "wal_level=logical"

gs_guc set -D ${PGDATA} -c "listen_addresses='*'"

gs_guc set -D ${PGDATA} -c "password_encryption_type=1"

给pg_hba.conf里增加(我这里的用户是gaussdb,根据实际情况修改)：

host replication gaussdb 0.0.0.0/0 sha256

host all all 0.0.0.0/0 sha256

openGauss的初始用户无法远程连接，必须新创建一个用户(我这里用户是gaussdb)：

create user gaussdb with login password "YOUR_PASSWORD";

grant all privileges to gaussdb;

为了方便测试，flink-sql-connector从源码的release-2.4分支编译，拷贝到flink的lib目录下后，按照文档操作，如果openGauss可以完全兼容postgres的话，我们可以得到和文档一样的结果，然而没有那么顺利。

问题定位

首先遇到的问题是版本判断小于postgres 9.4。对于这个问题，我们可以通过删掉版本检查的代码来绕过：

// flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java

private static void validateServerVersion(Statement statement) throws SQLException {

DatabaseMetaData metaData = statement.getConnection().getMetaData();

int majorVersion = metaData.getDatabaseMajorVersion();

int minorVersion = metaData.getDatabaseMinorVersion();

if (majorVersion < 9 || (majorVersion == 9 && minorVersion < 4)) {

// 删掉这个版本检查

// throw new SQLException("Cannot connect to a version of Postgres lower than 9.4");

}

考虑到postgres和openGauss的差异，我们把jdbc的依赖也一并修改掉避免难以预料的错误： flink-connector-postgres-cdc/pom.xml

org.opengauss

opengauss-jdbc

5.0.0

同时修改flink-sql-connector-postgres-cdc/pom.xml里的shade插件增加：

org.opengauss:*

执行mvn clean package -DskipTests重新打包后，将postgres connector的jar包拷贝到flink的lib目录下重新测试，这次出现了新的问题，flink job的日志报错显示connector在与openGauss建立连接后没有收到openGauss返回的任何网络包：

(e29effdca01963c9b8c2fa1e2d1c9888_605b35e407e90cda15ad084365733fdd_0_0) switched from RUNNING to FAILED with failure cause:

io.debezium.jdbc.JdbcConnectionException: [127.0.0.1:47500/ocalhost/127.0.0.1:5432] socket is not closed; Urgent packet sent to backend successfully; An I/O error occured while sending to the backend.detail:EOF Exception;

at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.initPublication(PostgresReplicationConnection.java:251) ~[flink-sql-connector-postgres-cdc-2.4-SNAPSHOT.jar:2.4-SNAPSHOT]

at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.createReplicationSlot(PostgresReplicationConnection.java:434) ~[flink-sql-connector-postgres-cdc-2.4-SNAPSHOT.jar:2.4-SNAPSHOT]

at io.debezium.connector.postgresql.PostgresConnectorTask.start(PostgresConnectorTask.java:137) ~[flink-sql-connector-postgres-cdc-2.4-SNAPSHOT.jar:2.4-SNAPSHOT]

at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:133) ~[flink-sql-connector-mysql-cdc-2.4-SNAPSHOT.jar:2.4-SNAPSHOT]

at io.debezium.embedded.EmbeddedEngine.run(EmbeddedEngine.java:760) ~[flink-sql-connector-mysql-cdc-2.4-SNAPSHOT.jar:2.4-SNAPSHOT]

at io.debezium.embedded.ConvertingEngineBuilder$2.run(ConvertingEngineBuilder.java:192) ~[flink-sql-connector-mysql-cdc-2.4-SNAPSHOT.jar:2.4-SNAPSHOT]

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[?:?]

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[?:?]

at java.lang.Thread.run(Thread.java:829) [?:?]

Caused by: org.postgresql.util.PSQLException: [127.0.0.1:47500/ocalhost/127.0.0.1:5432] socket is not closed; Urgent packet sent to backend successfully; An I/O error occured while sending to the backend.detail:EOF Exception;

at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:369) ~[postgresql.jar:?]

at org.postgresql.jdbc.PgStatement.runQueryExecutor(PgStatement.java:570) ~[postgresql.jar:?]

at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:547) ~[postgresql.jar:?]

at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:405) ~[postgresql.jar:?]

at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:347) ~[postgresql.jar:?]

at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:333) ~[postgresql.jar:?]

at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:310) ~[postgresql.jar:?]

at org.postgresql.jdbc.PgStatement.executeQuery(PgStatement.java:250) ~[postgresql.jar:?]

at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.initPublication(PostgresReplicationConnection.java:182) ~[flink-sql-connector-postgres-cdc-2.4-SNAPSHOT.jar:2.4-SNAPSHOT]

... 8 more

Caused by: java.io.EOFException: EOF Exception

at org.postgresql.core.PGStream.receiveChar(PGStream.java:309) ~[postgresql.jar:?]

at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2401) ~[postgresql.jar:?]

at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:341) ~[postgresql.jar:?]