{"id":273,"date":"2023-04-14T18:43:36","date_gmt":"2023-04-14T10:43:36","guid":{"rendered":"https:\/\/yhsbj.cn\/?p=273"},"modified":"2023-04-14T18:43:36","modified_gmt":"2023-04-14T10:43:36","slug":"flinkcdc%e5%90%8c%e6%ad%a5postgresql%e5%88%b0sqlserver","status":"publish","type":"post","link":"https:\/\/yhsbj.cn\/index.php\/2023\/04\/14\/273\/","title":{"rendered":"flinkcdc\u540c\u6b65PostgreSQL\u5230SQLServer"},"content":{"rendered":"\n

1.\u5fc5\u987b\u7528Linux\u7cfb\u7edf\uff0cWindows\u4e0b\u4f4e\u7248\u672cFlink\u6709Bug\uff0c\u9ad8\u7248\u672c\u7684API\u6539\u52a8\u592a\u5927\uff0c\u8d44\u6599\u592a\u5c11\u96be\u5f00\u53d1\u3002<\/p>\n\n\n\n

2.\u6211\u662f\u4f7f\u7528Docker\u7528Centos7\u955c\u50cf\uff0c\u7136\u540eFlink\u7248\u672c\u4e3a1.16.0\u3002https:\/\/archive.apache.org\/dist\/flink\/flink-1.16.1\/<\/a><\/p>\n\n\n\n

3.\u5b89\u88c5Java\uff0c\u53c2\u8003https:\/\/www.saoniuhuo.com\/article\/detail-92.html<\/a>\uff0c\u8fd9\u4e2a\u6559\u7a0b\u552f\u4e00\u4e0d\u597d\u7684\u5c31\u662f\u91cd\u542f\u540e\u73af\u5883\u53d8\u91cf\u5931\u6548\u3002<\/p>\n\n\n\n

4.SQLServer\u5f00\u542f\u5141\u8bb8\u8fdc\u7a0b\u8fde\u63a5\uff0c\u914d\u7f6e\u597d\u8bb0\u5f97\u91cd\u542f\u7535\u8111\uff0c\u91cd\u542f\u670d\u52a1\u4e0d\u884c\u3002<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

5.PostgreSQL\u6211\u4ece\u5b98\u7f51\u4e0b\u8f7d\u7684\u6700\u65b0\u7248\uff0c\u4e5f\u9700\u8981\u5f00\u542f\u5141\u8bb8\u8fdc\u7a0b\u8fde\u63a5\u3002<\/p>\n\n\n\n

6.\u5f00\u542f\u8fdc\u7a0b\u8fde\u63a5\u662f\u4e3a\u4e86\u8ba9Docker\u5bb9\u5668\u7684Centos\u53ef\u4ee5\u8bbf\u95ee\u5bbf\u4e3b\u7684\u6570\u636e\u5e93\u3002<\/p>\n\n\n\n

7.\u5206\u522b\u7ed9SQLServer\u548cPostgreSQL\u521b\u5efa\u8868\uff0cSQLServer\u7684id\u4e0d\u9700\u8981\u8bbe\u7f6e\u4e3b\u952e<\/span>\u3002<\/p>\n\n\n\n

\"\"
SQLServerbbibi’abiao\u8868<\/figcaption><\/figure>\n\n\n\n
\"\"
PostgreSQL\u8868\u8bbe\u8ba1\uff0c\u548cSQLServer\u662f\u4e00\u6837\u7684\u7c7b\u578b\uff0c\u4f46id\u6709\u4e3b\u952e\u548c\u81ea\u589e<\/figcaption><\/figure>\n\n\n\n

7.\u4fee\u6539PostgreSQL\u7684pg_hba.conf\u914d\u7f6e\u6587\u4ef6\uff0c\u591a\u52a0\u4e00\u6761\u4fe1\u606f\u8bbe\u7f6e\u5141\u8bb8\u8fdc\u7a0b\u8bbf\u95ee<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

8.\u4fee\u6539PostgreSQL\u7684postgresql.conf\u914d\u7f6e\u6587\u4ef6\uff0cFlinkCDC\u624d\u53ef\u4ee5\u76d1\u542c\u5230<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n
\"\"<\/figure>\n\n\n\n

9.IDEA\u521b\u5efaMaven\u9879\u76ee<\/p>\n\n\n\n

10.\u628a\u7528\u5230\u7684jar\u653e\u5230Flink\u7684lib\u76ee\u5f55\u4e0b\uff0c\u7136\u540eJava\u9879\u76ee\u6dfb\u52a0Jar\u7684\u5f15\u7528\u3002<\/p>\n\n\n\n

\"\"<\/figure>\n\n\n\n

11.\u5fc5\u987b\u8bf4\u660e\u4e00\u70b9\uff0cFlink\u9ad8\u548c\u4f4e\u7248\u672cAPI\u53d8\u5316\u5f88\u591a\uff0c\u5efa\u8baeJar\u7248\u672c\u548cFlink\u7248\u672c\u4fdd\u6301\u4e00\u81f4\uff01\uff01\uff01<\/span><\/strong><\/p>\n\n\n\n

12.\u521b\u5efaFlinkCdcDataStream.java\uff0c\u7f16\u5199Java\u4ee3\u7801\u76d1\u542cPostgreSQL\u8868\u7684\u53d8\u5316<\/p>\n\n\n\n

SourceFunction<String> sourceFunction = PostgreSQLSource.<String>builder()\n                    .hostname(\"192.168.0.148\").port(5432)\n                    .database(\"test_db_11\") \/\/ monitor postgres database\n                    .schemaList(\"public\")  \/\/ monitor inventory schema\n                    .tableList(\"public.test_table_11\") \/\/ monitor products table\n                    .decodingPluginName(\"pgoutput\")\n                    .slotName(\"t_table_slot\")\n                    .username(\"postgres\")\n                    .password(\"123\")\n                    .deserializer(new JsonDebeziumDeserializationSchema()) \/\/ converts SourceRecord to JSON String\n                    .build();\n\nStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();\nenv.enableCheckpointing(3000);\n\nenv.addSource(sourceFunction)\n   .addSink(new ClickHouseSink(\"test_table_mssql1\", \"id\"));\n\nenv.execute();<\/pre>\n\n\n\n
\"\"<\/figure>\n\n\n\n
13.\u521b\u5efaClickHouseSink.java\uff0c\u7f16\u5199\u76d1\u542c\u5230\u6570\u636e\u53d8\u5316\u540e\u540c\u6b65\u5230SQLServer\uff0c\u4ee3\u7801\u592a\u591a\u5c31\u4e0d\u622a\u56fe\uff0c\u4ee3\u7801\u653e\u5230\u4e0b\u9762\u5427<\/p>\n\n\n\n
package org.example;\n\nimport com.alibaba.fastjson2.JSON;\nimport com.alibaba.fastjson2.JSONObject;\nimport org.apache.flink.configuration.Configuration;\nimport org.apache.flink.streaming.api.functions.sink.RichSinkFunction;\n\nimport java.sql.Connection;\nimport java.sql.DriverManager;\nimport java.sql.PreparedStatement;\n\npublic class ClickHouseSink extends RichSinkFunction<String> {\n    private Connection connection;\n    private String targetTableName;\n    private String idName;\n\n    public ClickHouseSink(String targetTableName, String idName) {\n        this.targetTableName = targetTableName;\n        this.idName = idName;\n    }\n\n    @Override\n    public void open(Configuration parameters) throws Exception {\n        super.open(parameters);\n\n        try {\n            if (this.connection == null) {\n                connection = getConnection();\n            }\n        } catch (Exception e) {\n            System.out.println(e.getMessage());\n        }\n    }\n\n    @Override\n    public void close() {\n        try {\n            \/\/\u5173\u95ed\u8fde\u63a5\u548c\u91ca\u653e\u8d44\u6e90\n            if (connection != null) {\n                connection.close();\n            }\n        } catch (Exception e) {\n            System.out.println(e.getMessage());\n        }\n    }\n\n    @Override\n    public void invoke(String sql, Context context) {\n        JSONObject res = JSON.parseObject(sql);\n        System.out.println(res);\n        try {\n            JSONObject after = res.getJSONObject(\"after\");\n            JSONObject before = res.getJSONObject(\"before\");\n\n            switch (res.getString(\"op\")) {\n                case \"r\":\n                    System.out.println(\"\u8bfb\u53d6\u6570\u636e\");\n                    break;\n                case \"d\":\n                    System.out.println(\"\u5220\u9664\u6570\u636e\");\n                    if (before != null) {\n                        PreparedStatement ps = this.connection.prepareStatement(\"delete from \" + targetTableName + \" where \" + idName + \"=?\");\n                        ps.setInt(1, before.getInteger(idName));\n                        ps.execute();\n                        ps.close();\n                    }\n                    break;\n                case \"u\":\n                    System.out.println(\"\u66f4\u65b0\u6570\u636e\");\n                    PreparedStatement ps = this.connection.prepareStatement(\"update \" + targetTableName + \" set name=?, age=? where id = ?\");\n                    ps.setString(1, after.getString(\"name\"));\n                    ps.setInt(2, after.getInteger(\"age\"));\n                    ps.setInt(3, after.getInteger(\"id\"));\n                    ps.execute();\n                    ps.close();\n                    break;\n            }\n\n            if (res.getString(\"op\").equals(\"c\")) {\n                System.out.println(\"\u63d2\u5165\u6570\u636e\");\n                if (after != null) {\n                    PreparedStatement ps = this.connection.prepareStatement(\"insert into \" + targetTableName + \"(id, name, age) values(?,?,?)\");\n                    ps.setInt(1, after.getInteger(\"id\"));\n                    ps.setString(2, after.getString(\"name\"));\n                    ps.setInt(3, after.getInteger(\"age\"));\n                    ps.execute();\n                    ps.close();\n                }\n            }\n        } catch (Exception ex) {\n            System.out.println(\"\u5f02\u5e38\" + ex.getMessage());\n        }\n    }\n\n    private static Connection getConnection() {\n        Connection con = null;\n        try {\n            Class.forName(\"com.microsoft.sqlserver.jdbc.SQLServerDriver\").newInstance();\n            con = DriverManager.getConnection(\"jdbc:sqlserver:\/\/192.168.0.148:1433;database=test_db_mssql;user=test;password=123;encrypt=true;trustServerCertificate=true;loginTimeout=30;\");\n        } catch (Exception e) {\n            System.out.println(\"-----------sqlserver get connection has exception , msg = \" + e.getMessage());\n        }\n        return con;\n    }\n}<\/pre>\n\n\n\n
15.IDEA\u6dfb\u52a0\u5de5\u4ef6\uff0c\u5e76\u4e14\u9009\u62e9\u5165\u53e3\u7c7b<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
16.\u6784\u5efa\u5de5\u4ef6\u5f97\u5230\u4e00\u4e2aJar\u5305<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
\"\"<\/figure>\n\n\n\n
17.\u542f\u52a8Docker\u7aef\u53e3\u6620\u5c048081\u7136\u540e\u542f\u52a8Flink1.16.0\uff0c\u8bb0\u5f97\u9700\u8981\u63d0\u524d\u628aJar\u653e\u8fdbFlink\u7684lib\u4e0b\u9762\uff0c\u5426\u5219Flink\u65e0\u6cd5\u52a0\u8f7dJar\u4f1a\u4e0a\u4f20\u5931\u8d25<\/p>\n\n\n\n
18.\u63d0\u4ea4\u521a\u521a\u751f\u6210\u7684Jar\u5305\uff0c\u5e76\u4e14\u8fd0\u884c<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
\"\"<\/figure>\n\n\n\n
\"\"
\u8fd0\u884c\u6210\u529f<\/figcaption><\/figure>\n\n\n\n
\u81f3\u6b64\u5df2\u7ecf\u5b8c\u6210\u540c\u6b65\u529f\u80fd\u3002<\/p>\n\n\n\n
\u63a5\u4e0b\u6765\u770b\u4e0b\u6548\u679c\uff1a<\/p>\n\n\n\n
1.PostgreSQL\u63d2\u5165\u6570\u636e\uff0cSQLServer\u5b9e\u65f6\u540c\u6b65<\/p>\n\n\n\n
\"\"<\/figure>\n\n\n\n
\"\"<\/figure>\n\n\n\n
PostgreSQL\u9ed8\u8ba4\u60c5\u51b5\u65e0\u6cd5\u4fee\u6539\u8868\uff0c\u6211\u5c31\u4e0d\u91cd\u65b0\u5efa\u4e00\u4e2a\u8868\u4e86\u3002<\/p>\n\n\n\n
MySQL\u6d4b\u8bd5\u8fc7\u589e\u5220\u6539\u90fd\u662f\u5b9e\u65f6\u540c\u6b65\uff0c\u4ee3\u7801\u662f\u540c\u4e00\u4efd\u3002<\/p>\n\n\n\n
<\/p>\n","protected":false},"excerpt":{"rendered":"
1.\u5fc5\u987b\u7528Linux\u7cfb\u7edf\uff0cWindows\u4e0b\u4f4e\u7248\u672cFlink\u6709Bug\uff0c\u9ad8\u7248\u672c\u7684API\u6539\u52a8\u592a\u5927\uff0c\u8d44\u6599\u592a\u5c11\u96be\u5f00\u53d1\u3002  […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[1],"tags":[],"_links":{"self":[{"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/posts\/273"}],"collection":[{"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/comments?post=273"}],"version-history":[{"count":2,"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/posts\/273\/revisions"}],"predecessor-version":[{"id":292,"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/posts\/273\/revisions\/292"}],"wp:attachment":[{"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/media?parent=273"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/categories?post=273"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/yhsbj.cn\/index.php\/wp-json\/wp\/v2\/tags?post=273"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}