left join pgsql和mysql中的两张表生成新数据后放入mysql中

  • flink cdc pgsql
  • image.png

问题注意

Parallelism

  1. 只能设置为 1 , 要不然数据同步会出问题
    1. env.setParallelism_(_1_)_;
    2. SET 'parallelism.default' = '1'; -- optional: Flink's parallelism (1 by default)
    3. SET 'pipeline.max-parallelism' = '1'; -- optional: Flink's maximum parallelism

      submit job 时报错,但是任务提交却成功了

      ```java // sunmit job err tableEnv.executeSql(sourceUserDDL); tableEnv.executeSql(sourceClassesDDL); tableEnv.executeSql(sinkDDL); tableEnv.executeSql(transformDmlSQL);

// submit job success (真他妈神奇) TableResult tableResult = tableEnv.executeSql(sourceUserDDL); TableResult tableResult1 = tableEnv.executeSql(sourceClassesDDL); TableResult tableResult2 = tableEnv.executeSql(sinkDDL); TableResult tableResult3 = tableEnv.executeSql(transformDmlSQL);

  1. <a name="lsOYO"></a>
  2. # 建表语句
  3. ```sql
  4. # 源数据用户表 <MySQL>
  5. CREATE TABLE `user` (
  6.   `name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  7.   `id` int NOT NULL AUTO_INCREMENT,
  8.   `class_no` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
  9.   PRIMARY KEY (`id`)
  10. ) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
  13. # 源数据班级表 <postgres SQL>
  14. CREATE TABLE "public"."classes" (
  15.   "id" int4 NOT NULL DEFAULT nextval('classes_id_seq'::regclass),
  16.   "class_name" varchar(255) COLLATE "pg_catalog"."default" NOT NULL,
  17.   "class_no" varchar(255) COLLATE "pg_catalog"."default" NOT NULL,
  18.   CONSTRAINT "pk_port_forward" PRIMARY KEY ("id")
  19. );
  20. ALTER TABLE "public"."classes" 
  21.   OWNER TO "postgres";
  25. ## sink表<mysql>
  26. CREATE TABLE `user` (
  27.   `id` int NOT NULL,
  28.   `name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  29.   `class_name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  30.   PRIMARY KEY (`id`)
  31. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

Flink SQL准备

  1. CREATE TABLE mysql_user_binlog ( 
  2.   id INT NOT NULL,
  3.   name STRING, 
  4.   class_no STRING,
  5.   primary key (id) not enforced 
  6. ) WITH ( 
  7.   'connector' = 'mysql-cdc', 
  8.   'hostname' = '192.168.31.61', 
  9.   'port' = '3306', 
  10.   'username' = 'root', 
  11.   'password' = 'root', 
  12.   'database-name' = 'flink_sql', 
  13.   'table-name' = 'user' 
  14. );
  18.  CREATE TABLE postgres_class_binlog (
  19.    id INT NOT NULL,
  20.    class_name STRING,
  21.    class_no STRING,
  22.    primary key (id) not enforced
  23.  ) WITH ( 
  24.    'connector' = 'postgres-cdc',
  25.    'hostname' = '192.168.31.61',
  26.    'port' = '5432',
  27.    'username' = 'postgres',
  28.    'password' = 'root',
  29.    'database-name' = 'flink',
  30.    'schema-name' = 'public',
  31.    'table-name' = 'classes',
  32.    'decoding.plugin.name' = 'pgoutput'
  33.  );
  35. CREATE TABLE user_class_sink ( 
  36.   id INT, 
  37.   name STRING, 
  38.   class_name STRING, 
  39.   primary key (id) not enforced 
  40. ) WITH ( 
  41.   'connector' = 'jdbc', 
  42.   'driver' = 'com.mysql.cj.jdbc.Driver', 
  43.   'url' = 'jdbc:mysql://192.168.31.61:3306/flink?serverTimezone=Asia/Shanghai&useSSL=false',
  44.   'username' = 'root', 
  45.   'password' = 'root', 
  46.   'table-name' = 'user' 
  47. );
  49. SET 'pipeline.name' = 'PgSQLAndMySQL2MySQL';        
  51. SET 'parallelism.default' = '1'; 
  52. SET 'pipeline.max-parallelism' = '1';
  54. # sink                 
  55. insert into user_class_sink select mu.id, mu.name,pc.class_name from mysql_user_binlog as mu left join postgres_class_binlog as pc on mu.class_no = pc.class_no;

Maven依赖

  1. <dependencies>
  2.         <dependency>
  3.             <groupId>org.apache.flink</groupId>
  4.             <artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
  5.             <version>${flink.version}</version>
  6.         </dependency>
  8.         <!-- This dependency is provided, because it should not be packaged into the JAR file. -->
  9.         <dependency>
  10.             <groupId>org.apache.flink</groupId>
  11.             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  12.             <version>${flink.version}</version>
  13.             <scope>provided</scope>
  14.         </dependency>
  15.         <dependency>
  16.             <groupId>org.apache.flink</groupId>
  17.             <artifactId>flink-clients_${scala.binary.version}</artifactId>
  18.             <version>${flink.version}</version>
  19.             <scope>provided</scope>
  20.         </dependency>
  21.         <dependency>
  22.             <groupId>org.apache.logging.log4j</groupId>
  23.             <artifactId>log4j-slf4j-impl</artifactId>
  24.             <version>${log4j.version}</version>
  25.             <scope>runtime</scope>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.apache.logging.log4j</groupId>
  29.             <artifactId>log4j-api</artifactId>
  30.             <version>${log4j.version}</version>
  31.             <scope>runtime</scope>
  32.         </dependency>
  33.         <dependency>
  34.             <groupId>org.apache.logging.log4j</groupId>
  35.             <artifactId>log4j-core</artifactId>
  36.             <version>${log4j.version}</version>
  37.             <scope>runtime</scope>
  38.         </dependency>
  40.         <!--flink cdc connector -->
  41.         <dependency>
  42.             <groupId>com.ververica</groupId>
  43.             <!-- add the dependency matching your database -->
  44.             <artifactId>flink-sql-connector-mysql-cdc</artifactId>
  45.             <!-- the dependency is available only for stable releases. -->
  46.             <version>2.1.1</version>
  47.             <scope>provided</scope>
  48.         </dependency>
  49.       <!-- flink cdc pgsql -->
  50.         <dependency>
  51.             <groupId>com.ververica</groupId>
  52.             <artifactId>flink-sql-connector-postgres-cdc</artifactId>
  53.             <version>2.1.1</version>
  54.             <scope>provided</scope>
  55.         </dependency>
  56.         <dependency>
  57.             <groupId>org.apache.flink</groupId>
  58.             <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
  59.             <version>${flink.version}</version>
  60.             <scope>provided</scope>
  61.         </dependency>
  62.         <dependency>
  63.             <groupId>org.apache.flink</groupId>
  64.             <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
  65.             <version>${flink.version}</version>
  66.             <scope>provided</scope>
  67.         </dependency>
  69.     </dependencies>

JAVA代码

  1. package cn.tannn;
  4. import cn.tannn.constant.FlinSqlConstant;
  5. import org.apache.flink.streaming.api.CheckpointingMode;
  6. import org.apache.flink.streaming.api.environment.CheckpointConfig;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.table.api.EnvironmentSettings;
  9. import org.apache.flink.table.api.SqlDialect;
  10. import org.apache.flink.table.api.TableResult;
  11. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  13. /**
  14.  * post
  15.  */
  16. public class PgSQLAndMySQL2MySQL {
  18.     public static void main(String[] args) throws Exception {
  19.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.         env.enableCheckpointing(3000);
  21.         // 高级选项：
  22.         // 设置模式为exactly-once （这是默认值）
  23.         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  24.         // 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】
  25.         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(1500);
  26.         // 检查点必须在一分钟内完成，或者被丢弃【checkpoint的超时时间】
  27.         env.getCheckpointConfig().setCheckpointTimeout(60000);
  28.         // 同一时间只允许进行一个检查点
  29.         env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  30.         // 表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】
  31.         //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION:表示一旦Flink处理程序被cancel后，会保留Checkpoint数据，以便根据实际需要恢复到指定的Checkpoint
  32.         //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 表示一旦Flink处理程序被cancel后，会删除Checkpoint数据，只有job执行失败的时候才会保存checkpoint
  33.         env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  34.         env.setParallelism(1);
  36.         EnvironmentSettings Settings = EnvironmentSettings.newInstance()
  37.                 .useBlinkPlanner()
  38.                 .inStreamingMode()
  39.                 .build();
  41.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, Settings);
  42.         tableEnv.getConfig().getConfiguration().setString(FlinSqlConstant.JOB_NAME_KEY,"PgSQLAndMySQL2MySQL");
  43.         tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
  44.         // 数据源表 用户数据
  45.         String sourceUserDDL =  "CREATE TABLE mysql_user_binlog ( " +
  46.                                 " id INT NOT NULL, " +
  47.                                 " name STRING, " +
  48.                                 " class_no STRING, " +
  49.                                 " primary key (id) not enforced " +
  50.                                 ") WITH ( " +
  51.                                 " 'connector' = 'mysql-cdc', " +
  52.                                 " 'hostname' = '192.168.31.61', " +
  53.                                 " 'port' = '3306', " +
  54.                                 " 'username' = 'root', " +
  55.                                 " 'password' = 'root', " +
  56.                                 " 'database-name' = 'flink_sql', " +
  57.                                 " 'table-name' = 'user' " +
  58. //                                " , 'scan.startup.mode' = 'latest-offset' " + // 默认全量加增量
  59.                                 ")  ";
  60.         // 数据源表 班级数据 pgsql
  61.         String sourceClassesDDL =  "CREATE TABLE postgres_class_binlog ( " +
  62.                 " id INT NOT NULL, " +
  63.                 " class_name STRING, " +
  64.                 " class_no STRING, " +
  65.                 " primary key (id) not enforced " +
  66.                 ") WITH ( " +
  67.                 " 'connector' = 'postgres-cdc', " +
  68.                 " 'hostname' = '192.168.31.61', " +
  69.                 " 'port' = '5432', " +
  70.                 " 'username' = 'postgres', " +
  71.                 " 'password' = 'root', " +
  72.                 " 'database-name' = 'flink', " +
  73.                 " 'schema-name' = 'public', " +
  74.                 " 'table-name' = 'classes', " +
  75.                 " 'decoding.plugin.name' = 'pgoutput' " +
  76. //                " 'debezium.slot.name' = '***' " +
  77. //                                " , 'scan.startup.mode' = 'latest-offset' " + // 默认全量加增量
  78.                 ")  ";
  79.         // 输出目标表 聚合用户跟class信息
  80.         String sinkDDL =
  81.                 "CREATE TABLE user_class_sink ( " +
  82.                         " id INT, " +
  83.                         " name STRING, " +
  84.                         " class_name STRING, " +
  85.                         " primary key (id) not enforced " +
  86.                         ") WITH ( " +
  87.                         " 'connector' = 'jdbc', " +
  88.                         " 'driver' = 'com.mysql.cj.jdbc.Driver', " +
  89.                         " 'url' = 'jdbc:mysql://192.168.31.61:3306/flink?serverTimezone=Asia/Shanghai&useSSL=false'," +
  90.                         " 'username' = 'root', " +
  91.                         " 'password' = 'root', " +
  92.                         " 'table-name' = 'user' " +
  93.                         ")";
  95.         // 简单的聚合处理
  96.         String transformDmlSQL =  "insert into user_class_sink select mu.id, mu.name,pc.class_name " +
  97.                 " from mysql_user_binlog as mu left join postgres_class_binlog as pc" +
  98.                 " on mu.class_no = pc.class_no";
  100.          TableResult tableResult = tableEnv.executeSql(sourceUserDDL);
  101.         TableResult tableResult1 = tableEnv.executeSql(sourceClassesDDL);
  102.         TableResult tableResult2 = tableEnv.executeSql(sinkDDL);
  103.         TableResult tableResult3 = tableEnv.executeSql(transformDmlSQL);
  105.     }
  107. }