FlinkSQL MySQL CDC

雨点打透心脏的1/2处 2021-09-24 21:10 337阅读 0赞

[flink-cdc-connectors][]

package com.sm.job
    
    import com.sm.function.udf.{ParseDeviceCode, ParsePopularize}
    import com.sm.utils.FlinkUtils
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
    import org.apache.log4j.Level
    import org.slf4j.LoggerFactory
    
    /**
      * flink mysql CDC
      * 实时同步维表
      *
      * create by LiuJinHe 2020/10/22
      */
    object FlinkMySqlCdc {
      private var logger: org.slf4j.Logger = _
    
      def main(args: Array[String]): Unit = {
        logger = LoggerFactory.getLogger(this.getClass.getSimpleName)
        org.apache.log4j.Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
        org.apache.log4j.Logger.getLogger("org.apache").setLevel(Level.INFO)
        org.apache.log4j.Logger.getLogger("io.debezium").setLevel(Level.INFO)
    
        // 初始化 stream 环境
        // 本地测试，需要 flink-runtime-web 依赖
        val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
        //    val env = StreamExecutionEnvironment.getExecutionEnvironment
    
        // 失败重启,固定间隔,每隔3秒重启1次,总尝试重启10次
        //    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, 3))
    
        // 本地测试线程 1
        //    env.setParallelism(1)
    
        // 事件处理的时间，由系统时间决定
        //    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    
        // 创建 streamTable 环境
        val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, FlinkUtils.getSettings)
        //    tableConfig.setIdleStateRetentionTime(Time.hours(12), Time.hours(24))
    
        // checkpoint 设置
        // val tableConfig = tableEnv.getConfig.getConfiguration
        // 开启checkpoint
        tableConfig.set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE)
        // checkpoint的超时时间周期，1 分钟做一次checkpoint, 每次checkpoint 完成后 sink 才会执行
        tableConfig.set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(60))
        // checkpoint的超时时间, 检查点一分钟内没有完成将被丢弃
        // tableConfig.set(ExecutionCheckpointingOptions.CHECKPOINTING_TIMEOUT, Duration.ofSeconds(60))
        // checkpoint 最小间隔，两个检查点之间至少间隔 30 秒
        // tableConfig.set(ExecutionCheckpointingOptions.MIN_PAUSE_BETWEEN_CHECKPOINTS, Duration.ofSeconds(30))
        // 同一时间只允许进行一个检查点
        // tableConfig.set(ExecutionCheckpointingOptions.MAX_CONCURRENT_CHECKPOINTS, Integer.valueOf(1))
        // 手动cancel时是否保留checkpoint
        // tableConfig.set(ExecutionCheckpointingOptions.EXTERNALIZED_CHECKPOINT,
          CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    
        // 设置状态的最小空闲时间和最大的空闲时间
        // tableEnv.getConfig.setIdleStateRetentionTime(Time.hours(12), Time.hours(24))
    
        // Catalog
        //    val memoryCatalog = new GenericInMemoryCatalog("kafkaSourceTable", "memory")
    
        // 删除null数据
        //    tableEnv.getConfig.getConfiguration.setString("table.exec.sink.not-null-enforcer", "drop")
    
        val packageSourceSql =
        """
          |create table packageSourceTable (
          |  ID int,
          |  PACKAGE_ID int,
          |  GAME_ID int,
          |  CHANNEL_ID int,
          |  PLATFORM int,
          |  POPULARIZE int,
          |  COMPANY_ID int,
          |  ADD_TIME as localtimestamp,
          |  proctime as proctime()
          |) with (
          | 'connector' = 'mysql-cdc',
          | 'hostname' = 'xxx',
          | 'port' = '3306',
          | 'username' = 'xxx',
          | 'password' = 'xxx',
          | 'database-name' = 'game_platform',
          | 'table-name' = 'packages',
          | 'server-time-zone' = 'Asia/Shanghai'
          |)
        """.stripMargin
    
        val printSinkSql =
        """
          |create table printSinkTable (
          |  ID int,
          |  PACKAGE_ID int,
          |  GAME_ID int,
          |  CHANNEL_ID int,
          |  PLATFORM int,
          |  POPULARIZE int,
          |  COMPANY_ID int,
          |  ADD_TIME timestamp
          |) with (
          | 'connector' = 'print'
          |)
        """.stripMargin
    
        val insertSql =
        """
          |insert into printSinkTable
          |select
          |  ID,
          |  PACKAGE_ID,
          |  GAME_ID,
          |  CHANNEL_ID,
          |  PLATFORM,
          |  POPULARIZE,
          |  COMPANY_ID,
          |  ADD_TIME
          |from packageSourceTable
        """.stripMargin
    
        tableEnv.executeSql(packageSourceSql)
        tableEnv.executeSql(printSinkSql)
        tableEnv.executeSql(insertSql)
      }
    }

table API

val mysqlSource = MySQLSource.builder()
          .hostname("cdh-slave04")
          .port(3306)
          .username("root")
          .password("123456")
          .databaseList("game_platfrom")
          .tableList("game_platfrom.games")
          .deserializer(new StringDebeziumDeserializationSchema)
          .build()
    
        env.addSource(mysqlSource).print()
    
        env.execute("flink mysql cdc")

[flink-cdc-connectors]: https://github.com/ververica/flink-cdc-connectors/tree/release-1.1