怎样利用Flink连接openGauss数据库（flink-cdc-connector） - IT评测·应用市场-qidao123.com

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)

复制代码

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'
);

复制代码

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"

复制代码

host replication gaussdb 0.0.0.0/0 sha256
host all all 0.0.0.0/0 sha256

复制代码

create user gaussdb with login password "YOUR_PASSWORD";
grant all privileges to gaussdb;

复制代码

// 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");
}
}

复制代码

<dependency>
<groupId>org.opengauss</groupId>
<artifactId>opengauss-jdbc</artifactId>
<version>5.0.0</version>
</dependency>

复制代码

<include>org.opengauss:*</include>

复制代码

(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:?]
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

复制代码

walsender thread started
No message received from standby for maximum time
walsender thread shut down

复制代码

gs_guc set -D ${PGDATA} -c "log_statement = 'all'"

复制代码

this.jdbcConnection = jdbcConnection;

复制代码

Connection conn = jdbcConnection.connection();

复制代码

// NOTE: it requires ANALYZE or VACUUM to be run first in PostgreSQL.
final String query =
String.format(
- "SELECT reltuples::bigint FROM pg_class WHERE oid = to_regclass('%s')",
+ "SELECT reltuples::bigint FROM pg_class WHERE oid = '%s'::regclass",
tableId.toString());

复制代码

private Set<TableId> determineCapturedTables() throws Exception {
Set<TableId> allTableIds =
- this.connect()
- .readTableNames(
- pgConnection().getCatalog(), null, null, new String[] {"TABLE"});
+ this.jdbcConnection.readTableNames(null, null, null, new String[] {"TABLE"});

复制代码

@@ -71,14 +71,12 @@ public class PostgresDialect extends AbstractDialect {
.collect(Collectors.joining(", "));
String updateClause =
Arrays.stream(fieldNames)
+ .filter(s -> !uniqueColumns.contains(s))
.map(f -> quoteIdentifier(f) + "=EXCLUDED." + quoteIdentifier(f))
.collect(Collectors.joining(", "));
return Optional.of(
getInsertIntoStatement(tableName, fieldNames)
- + " ON CONFLICT ("
- + uniqueColumns
- + ")"
- + " DO UPDATE SET "
+ + " ON DUPLICATE KEY UPDATE "
+ updateClause);
}

复制代码

<dependency>
<groupId>xxx.xxx</groupId>
<artifactId>xxx</artifactId>
<version>your_version</version>
<scope>system</scope>
<systemPath>/path/to/jdbc.jar</systemPath>
</dependency>

复制代码

select pg_create_logical_replication_slot('your_slot_name','pgoutput');
create publication dbz_publication for all tables;

复制代码

commit 3d8924b37ed3d25764788b63ada1f48e00a97c4eAuthor: WangYuxuan <wang.yuxuan.xinyu@foxmail.com>Date: Wed Aug 23 16:23:13 2023 +0800 example commitdiff --git a/flink-connector-postgres-cdc/pom.xml b/flink-connector-postgres-cdc/pom.xmlindex a9886dc8..d7637615 100644--- a/flink-connector-postgres-cdc/pom.xml+++ b/flink-connector-postgres-cdc/pom.xml@@ -67,12 +67,18 @@ under the License. </exclusions> </dependency> +  <groupId>org.postgresql</groupId> <artifactId>postgresql</artifactId> <version>42.5.1</version> </dependency>+ -->+ <dependency>+ <groupId>org.opengauss</groupId>+ <artifactId>opengauss-jdbc</artifactId>+ <version>5.0.0</version>+ </dependency>  @@ -168,4 +174,4 @@ under the License. </dependencies> -</project>\ No newline at end of file+</project>diff --git a/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/utils/PostgresQueryUtils.java b/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/utils/PostgresQueryUtils.javaindex 3d3fc996..e30443a8 100644--- a/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/utils/PostgresQueryUtils.java+++ b/flink-connector-postgres-cdc/src/main/java/com/ververica/cdc/connectors/postgres/source/utils/PostgresQueryUtils.java@@ -67,7 +67,7 @@ public class PostgresQueryUtils { // NOTE: it requires ANALYZE or VACUUM to be run first in PostgreSQL.
final String query =
String.format(
- "SELECT reltuples::bigint FROM pg_class WHERE oid = to_regclass('%s')",
+ "SELECT reltuples::bigint FROM pg_class WHERE oid = '%s'::regclass",
tableId.toString());
return jdbc.queryAndMap(diff --git a/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java b/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.javaindex e728f3b4..1c21dfd6 100644--- a/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java+++ b/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresConnection.java@@ -636,7 +636,8 @@ public class PostgresConnection extends JdbcConnection { 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");+ // throw new SQLException("Cannot connect to a version of Postgres lower than 9.4");+ LOGGER.warn("wyx"); } } diff --git a/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java b/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.javaindex 364cb59b..eeb6a4e8 100644--- a/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java+++ b/flink-connector-postgres-cdc/src/main/java/io/debezium/connector/postgresql/connection/PostgresReplicationConnection.java@@ -74,6 +74,7 @@ public class PostgresReplicationConnection extends JdbcConnection implements Rep private final MessageDecoder messageDecoder; private final TypeRegistry typeRegistry; private final Properties streamParams;+ private final PostgresConnection jdbcConnection; private Lsn defaultStartingPos; private SlotCreationResult slotCreationInfo;@@ -136,6 +137,7 @@ public class PostgresReplicationConnection extends JdbcConnection implements Rep this.statusUpdateInterval = statusUpdateInterval; this.messageDecoder = plugin.messageDecoder(new MessageDecoderContext(config, schema), jdbcConnection);+ this.jdbcConnection = jdbcConnection;
this.typeRegistry = typeRegistry; this.streamParams = streamParams; this.slotCreationInfo = null;@@ -171,7 +173,7 @@ public class PostgresReplicationConnection extends JdbcConnection implements Rep LOGGER.info("Initializing PgOutput logical decoder publication"); try { // Unless the autocommit is disabled the SELECT publication query will stay running- Connection conn = pgConnection();+ Connection conn = jdbcConnection.connection();
conn.setAutoCommit(false); String selectPublication =@@ -255,9 +257,7 @@ public class PostgresReplicationConnection extends JdbcConnection implements Rep private Set<TableId> determineCapturedTables() throws Exception {
Set<TableId> allTableIds =
- this.connect()
- .readTableNames(
- pgConnection().getCatalog(), null, null, new String[] {"TABLE"});
+ this.jdbcConnection.readTableNames(null, null, null, new String[] {"TABLE"});
Set<TableId> capturedTables = new HashSet<>(); diff --git a/flink-sql-connector-postgres-cdc/pom.xml b/flink-sql-connector-postgres-cdc/pom.xmlindex 8beb8bc6..c8974e0e 100644--- a/flink-sql-connector-postgres-cdc/pom.xml+++ b/flink-sql-connector-postgres-cdc/pom.xml@@ -65,6 +65,7 @@ under the License. <include>com.fasterxml.*:*</include>  <include>org.apache.flink:flink-shaded-guava</include>+ <include>org.opengauss:*</include>
</includes> </artifactSet> <filters>@@ -107,4 +108,4 @@ under the License. </plugin> </plugins> </build>-</project>\ No newline at end of file+</project>

复制代码