# 更改wal日志方式为logical
wal_level = logical # minimal, replica, or logical

# 更改solts最大数量（默认值为10），flink-cdc默认一张表占用一个slots
max_replication_slots = 20 # max number of replication slots

# 更改wal发送最大进程数（默认值为10），这个值和上面的solts设置一样
max_wal_senders = 20 # max number of walsender processes
# 中断那些停止活动超过指定毫秒数的复制连接，可以适当设置大一点（默认60s）
wal_sender_timeout = 180s # in milliseconds; 0 disable

-- pg新建用户
CREATE USER user WITH PASSWORD 'pwd';

-- 给用户复制流权限
ALTER ROLE user replication;

-- 给用户登录数据库权限
grant CONNECT ON DATABASE test to user;

-- 把当前库public下所有表查询权限赋给用户
GRANT SELECT ON ALL TABLES IN SCHEMA public TO user;

-- 设置发布为true
update pg_publication set puballtables=true where pubname is not null; -- 把所有表进行发布
CREATE PUBLICATION dbz_publication FOR ALL TABLES; -- 查询哪些表已经发布 select * from pg_publication_tables;

-- 更改复制标识包含更新和删除之前值
ALTER TABLE test0425 REPLICA IDENTITY FULL;
-- 查看复制标识（为f标识说明设置成功）
select relreplident from pg_class where relname='test0425';

-- pg新建用户
CREATE USER ODPS_ETL WITH PASSWORD 'odpsETL@2021'; -- 给用户复制流权限
ALTER ROLE ODPS_ETL replication; -- 给用户数据库权限
grant CONNECT ON DATABASE test to ODPS_ETL; -- 设置发布开关
update pg_publication set puballtables=true where pubname is not null; -- 把所有表进行发布
CREATE PUBLICATION dbz_publication FOR ALL TABLES; -- 查询哪些表已经发布 select * from pg_publication_tables; -- 给表查询权限
grant select on TABLE aa to ODPS_ETL; -- 给用户读写权限
grant select,insert,update,delete ON  ALL TABLES IN SCHEMA public to bd_test; -- 把当前库所有表查询权限赋给用户
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ODPS_ETL; -- 把当前库以后新建的表查询权限赋给用户
alter default privileges in schema public grant select on tables to ODPS_ETL; -- 更改复制标识包含更新和删除之前值
ALTER TABLE test0425 REPLICA IDENTITY FULL; -- 查看复制标识 select relreplident from pg_class where relname='test0425'; -- 查看solt使用情况
SELECT * FROM pg_replication_slots; -- 删除solt
SELECT pg_drop_replication_slot('zd_org_goods_solt'); -- 查询用户当前连接数 select usename, count(*) from pg_stat_activity group by usename order by count(*) desc; -- 设置用户最大连接数
alter role odps_etl connection limit 200;

        <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-scala_2.11</artifactId> <version>1.13.0</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-scala_2.11</artifactId> <version>1.13.0</version> </dependency> <dependency> <groupId>com.alibaba.ververica</groupId> <artifactId>flink-connector-postgres-cdc</artifactId> <version>1.1.0</version> </dependency>

package flinkTest.connect; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableResult; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; public class PgsqlToMysqlTest { public static void main(String[] args) { //设置flink表环境变量 EnvironmentSettings fsSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build(); //获取flink流环境变量 StreamExecutionEnvironment exeEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        exeEnv.setParallelism(1); //表执行环境 StreamTableEnvironment tableEnv = StreamTableEnvironment.create(exeEnv, fsSettings); //拼接souceDLL String sourceDDL =
                "CREATE TABLE pgsql_source (\n" +
                        " id int,\n" +
                        " name STRING,\n" +
                        " py_code STRING,\n" +
                        " seq_no int,\n" +
                        " description STRING\n" +
                        ") WITH (\n" +
                        " 'connector' = 'postgres-cdc',\n" +
                        " 'hostname' = '***',\n" +
                        " 'port' = '5432',\n" +
                        " 'username' = 'bd_test',\n" +
                        " 'password' = '***',\n" +
                        " 'database-name' = 'bd_test',\n" +
                        " 'schema-name' = 'public',\n" +
                        " 'debezium.snapshot.mode' = 'never',\n" +
                        " 'decoding.plugin.name' = 'pgoutput',\n" +
                        " 'debezium.slot.name' = 'test',\n" +
                        " 'table-name' = 'test'\n" +
                        ")";

        String sinkDDL =
                "CREATE TABLE mysql_sink (\n" +
                        " id int,\n" +
                        " name STRING,\n" +
                        " py_code STRING,\n" +
                        " seq_no int,\n" +
                        " description STRING,\n" +
                        " PRIMARY KEY (id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        " 'connector' = 'jdbc',\n" +
                        " 'url' = 'jdbc:mysql://ip:3306/test_db?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=UTF-8',\n" +
                        " 'username' = 'bd_test',\n" +
                        " 'password' = '***',\n" +
                        " 'table-name' = 'test'\n" +
                        ")";

        String transformSQL =
                "INSERT INTO mysql_sink " +
                        "SELECT id,name,py_code,seq_no,description " +
                        "FROM pgsql_source"; //执行source表ddl  tableEnv.executeSql(sourceDDL); //执行sink表ddl  tableEnv.executeSql(sinkDDL); //执行逻辑sql语句 TableResult tableResult = tableEnv.executeSql(transformSQL); //控制塔输出 // tableResult.print();  }
}

-- pgsql表结构 CREATE TABLE "public"."test" (
  "id" int4 NOT NULL,
  "name" varchar(50) COLLATE "pg_catalog"."default" NOT NULL,
  "py_code" varchar(50) COLLATE "pg_catalog"."default",
  "seq_no" int4 NOT NULL,
  "description" varchar(200) COLLATE "pg_catalog"."default", CONSTRAINT "pk_zd_business_type" PRIMARY KEY ("id")
)
; -- mysql表结构 CREATE TABLE `test` (
  `id` int(11) NOT NULL DEFAULT '0' COMMENT 'ID',
  `name` varchar(50) DEFAULT NULL COMMENT '名称',
  `py_code` varchar(50) DEFAULT NULL COMMENT '助记码',
  `seq_no` int(11) DEFAULT NULL COMMENT '排序',
  `description` varchar(200) DEFAULT NULL COMMENT '备注', PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;