devbox
IT小白
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Flink CDC2.4 整库实时同步MySql 到Doris_flick mysql doris

作者：IT小白 | 2024-04-10 12:21:49

赞

踩

flick mysql doris

环境

        Flink 1.15.4 

实现原因

        目前有很多工具都支持无代码实现Mysql -> Doris 的实时同步

        如：SlectDB 已发布的功能包

                Dinky SeaTunnel TIS 等等

         不过好多要么不支持表结构变动，要不不支持多sink，我们的业务必须支持对表结构的实时级变动，因为会对表字段级别的修改，字段类型更改，字段名字更改删除添加等

        所以要支持整库同步且又要表结构的实时变动就要自己写

                

所需jar

        flink-doris-connector-1.15-1.4.0.jar  -- 实现一键万表同步

        flink-sql-connector-mysql-cdc-2.4.0.jar --包含所有相关依赖，无需在导入debezium、cdc等等

流程

        1、脚本创建库表

        2、同步表结构程序  

        3、Flink cdc 程序

对比第一版本:使用 Flink CDC 实现 MySQL 数据,表结构实时入 Apache Doris 效率有所提升

        首次同步时keyby 后开窗聚合导致数据倾斜

        聚合数据有字符串拼接改为JsonArray 避免聚合导致背压，字符串在数据量较大时拼接效率太低

Flink cdc 代码

        1、FlinkSingleSync.scala

        

  1. package com.zbkj.sync
  2.  
  3. import com.alibaba.fastjson2.{JSON, JSONObject,JSONArray}
  4. import com.ververica.cdc.connectors.mysql.source.MySqlSource
  5. import com.ververica.cdc.connectors.mysql.table.StartupOptions
  6. import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.json.JsonConverterConfig
  7. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema
  8. import com.zbkj.util.SinkBuilder.getKafkaSink
  9. import com.zbkj.util._
  10. import org.apache.flink.api.common.eventtime.WatermarkStrategy
  11. import org.apache.flink.api.common.restartstrategy.RestartStrategies
  12. import org.apache.flink.api.java.utils.ParameterTool
  13. import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
  14. import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
  15. import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
  16. import org.apache.flink.streaming.api.windowing.time.Time
  17. import org.apache.flink.streaming.api.scala._
  18.  
  19. import java.util.Properties
  20.  
  21. object FlinkSingleSync {
  22.  
  23.   PropertiesManager.initUtil()
  24.   val props: PropertiesUtil = PropertiesManager.getUtil
  25.  
  26.   def main(args: Array[String]): Unit = {
  27.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  28.     // 并行度
  29.     env.setParallelism(props.parallelism)
  30.     env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
  31.     
  32.     val parameters: ParameterTool = ParameterTool.fromArgs(args)
  33.     val memberID = parameters.getInt("memberID", 0)
  34.     val source = parameters.get("source", "")
  35.     val log = parameters.getBoolean("log", true)
  36.     if (memberID == 0) {
  37.       sys.exit(0)
  38.     }
  39.     val thisMember = "ttk_member_%d".format(memberID)
  40.     val jobName = "Sync Member %d".format(memberID)
  41.     val syncTopic = "sync_data_%d".format(memberID)
  42.     println(syncTopic)
  43.     val sourceFormat = SourceFormat.sourceFormat(source)
  44.  
  45.     env.setParallelism(4)
  46.     /**
  47.      * checkpoint的相关设置 */
  48.     // 启用检查点，指定触发checkpoint的时间间隔（单位：毫秒，默认500毫秒），默认情况是不开启的
  49.     env.enableCheckpointing(1000L, CheckpointingMode.EXACTLY_ONCE)
  50.     // 设定Checkpoint超时时间，默认为10分钟
  51.     env.getCheckpointConfig.setCheckpointTimeout(600000)
  52.  
  53.     /**
  54.      * 设置检查点路径 */
  55.     env.getCheckpointConfig.setCheckpointStorage("file:///data/flink-checkpoints/sync/%d".format(memberID))
  56.  
  57.     /** 设定两个Checkpoint之间的最小时间间隔，防止出现例如状态数据过大而导致Checkpoint执行时间过长，从而导致Checkpoint积压过多
  58.      * 最终Flink应用密切触发Checkpoint操作，会占用了大量计算资源而影响到整个应用的性能（单位：毫秒） */
  59.     env.getCheckpointConfig.setMinPauseBetweenCheckpoints(60000)
  60.     // 默认情况下，只有一个检查点可以运行
  61.     // 根据用户指定的数量可以同时触发多个Checkpoint，进而提升Checkpoint整体的效率
  62.     //env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
  63.     /** 外部检查点
  64.      * 不会在任务正常停止的过程中清理掉检查点数据，而是会一直保存在外部系统介质中，另外也可以通过从外部检查点中对任务进行恢复 */
  65.     env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
  66.  
  67.  
  68.     //    env.getCheckpointConfig.setPreferCheckpointForRecovery(true)
  69.     // 设置可以允许的checkpoint失败数
  70.     env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)
  71.     //设置可容忍的检查点失败数，默认值为0表示不允许容忍任何检查点失败
  72.     env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)
  73.     env.disableOperatorChaining()
  74.  
  75.     /**
  76.      * 重启策略的配置
  77.      * 重启3次，每次失败后等待10000毫秒
  78.      */
  79.     env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 30000L))
  80.  
  81.     val dataBaseList = thisMember
  82.     var tableList = thisMember + ".*"
  83.     if (!log) {
  84.       tableList = "lb_crm_customer_log|.*(?<!_log)$"
  85.     }
  86.  
  87.     val dorisStreamLoad = new DorisStreamLoad2(props)
  88.  
  89.     // numeric 类型转换
  90.     val customConverterConfigs = new java.util.HashMap[String, Object] {
  91.       put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric")
  92.     }
  93.     /**
  94.      * mysql source for doris */
  95.     println(dataBaseList, tableList)
  96.     val debeziumProps = new Properties()
  97.     debeziumProps.setProperty("debezium.snapshot.mode","never")
  98.     val mysqlSource = MySqlSource.builder[String]()
  99.       .hostname(sourceFormat.getString("sourceHost"))
  100.       .port(sourceFormat.getIntValue("sourcePort"))
  101.       .databaseList(dataBaseList)
  102.       //^((?!lb_admin_log|lb_bugs).)*$
  103.       // lb_admin_log、lb_bugs为不需要同步表
  104.       .tableList(props.regular_expression)
  105.       .username(sourceFormat.getString("sourceUsername"))
  106.       .password(sourceFormat.getString("sourcePassword"))
  107.       .debeziumProperties(debeziumProps)
  108.       // 全量读取
  109.       .startupOptions(StartupOptions.initial())
  110.       .includeSchemaChanges(true)
  111.       // 发现新表，加入同步任务，需要在tableList中配置
  112.       .scanNewlyAddedTableEnabled(true)
  113.       .deserializer(new JsonDebeziumDeserializationSchema(false, customConverterConfigs)).build()
  114.  
  115.     val streamSource: DataStream[JSONObject] = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
  116.       .map(line => JSON.parseObject(line)).setParallelism(4)
  117.  
  118.     val DDLSqlStream: DataStream[JSONObject] = streamSource.filter(line => !line.containsKey("op")).uid("ddlSqlStream")
  119.     val DMLStream: DataStream[JSONObject] = streamSource.filter(line => line.containsKey("op")).uid("dmlStream")
  120.     /**
  121.      * 首次全量同步时 时间窗口内几乎为一个表数据，此时下面操作会数据倾斜
  122.      * 在binLogETLOne 中对表加随机数后缀 使其均匀分布
  123.      * 聚合操作之后再将tableName转换为实际表
  124.      */
  125.     val DMLDataStream = FlinkCDCSyncETL.binLogETLOne(DMLStream)
  126.     val keyByDMLDataStream:DataStream[(String, String, String, JSONArray)] = DMLDataStream.keyBy(keys => (keys._1, keys._2, keys._3))
  127.       .timeWindow(Time.milliseconds(props.window_time_milliseconds))
  128.       .reduce((itemFirst, itemSecond) => (itemFirst._1, itemFirst._2, itemFirst._3,combineJsonArray(itemFirst._4,itemSecond._4)))
  129.       .map(line=>(line._1,line._2,line._3.split("-")(0),line._4))
  130.       .name("分组聚合").uid("keyBy")
  131.  
  132.  
  133.     keyByDMLDataStream.addSink(new SinkDoris(dorisStreamLoad)).name("数据写入Doris").uid("SinkDoris").setParallelism(4)
  134.  
  135.     val DDLKafkaSink=getKafkaSink("schema_change")
  136.     DDLSqlStream.map(jsObj => jsObj.toJSONString()).sinkTo(DDLKafkaSink).name("同步DDL入Kafka").uid("SinkDDLKafka")
  137.     
  138.     val kafkaSink=getKafkaSink(syncTopic)
  139.     keyByDMLDataStream.map(line=>(line._2,line._3,1)).filter(!_._2.endsWith("_sql"))
  140.       .keyBy(keys => (keys._1, keys._2))
  141.       .window(TumblingProcessingTimeWindows.of(Time.seconds(1))).sum(2)
  142.       .map(line =>{
  143.         val json = new JSONObject()
  144.         json.put("member_id", line._1)
  145.         json.put("table", line._2)
  146.         json.toJSONString()
  147.       }).sinkTo(kafkaSink).name("同步数据库表入Kafka").uid("syncDataTableToKafka")
  148.  
  149.     env.execute(jobName)
  150.     
  151.   }
  152.   
  153.   def combineJsonArray(jsr1:JSONArray,jsr2:JSONArray): JSONArray ={
  154.     jsr1.addAll(jsr2)
  155.     jsr1
  156.   }
  157.  
  158. }

2.FlinkCDCSyncETL.scala

  1. package com.zbkj.util
  2.  
  3. import com.alibaba.fastjson2.{JSON, JSONArray, JSONObject}
  4. import org.apache.flink.api.scala.createTypeInformation
  5. import org.apache.flink.streaming.api.scala.DataStream
  6.  
  7. import java.util.Random
  8.  
  9. object FlinkCDCSyncETL {
  10.  
  11.   def binLogETLOne(dataStreamSource: DataStream[JSONObject]): DataStream[(String, String, String, JSONArray)] = {
  12.     /**
  13.      * 根据不同日志类型 匹配load doris方式
  14.      */
  15.     val tupleData: DataStream[(String, String, String, JSONArray)] = dataStreamSource.map(line => {
  16.       var data: JSONObject = new JSONObject()
  17.       var jsr: JSONArray = new JSONArray()
  18.       var mergeType = "APPEND"
  19.       val source = line.getJSONObject("source")
  20.       val db = source.getString("db")
  21.       val table = source.getString("table")
  22.       val op=line.getString("op")
  23.       if ("d" == op) {
  24.         data = line.getJSONObject("before")
  25.         mergeType = "DELETE"
  26.       } else if ("u" == op) {
  27.         data = line.getJSONObject("after")
  28.         mergeType = "APPEND"
  29.       } else if ("c" == op) {
  30.         data = line.getJSONObject("after")
  31.       } else if ("r" == op) {
  32.         data = line.getJSONObject("after")
  33.         mergeType = "APPEND"
  34.       }
  35.       jsr.add(data)
  36.       Tuple4(mergeType, db, table+ "-" + new Random().nextInt(4), jsr)
  37.     })
  38.     tupleData
  39.   }
  40.  
  41.  
  42.  
  43. }

3.DorisStreamLoad2.scala

  1. package com.zbkj.util
  2.  
  3. import org.apache.doris.flink.exception.StreamLoadException
  4. import org.apache.doris.flink.sink.HttpPutBuilder
  5. import org.apache.http.client.methods.CloseableHttpResponse
  6. import org.apache.http.entity.StringEntity
  7. import org.apache.http.impl.client.{DefaultRedirectStrategy, HttpClientBuilder, HttpClients}
  8. import org.apache.http.util.EntityUtils
  9. import org.slf4j.{Logger, LoggerFactory}
  10.  
  11. import java.util.{Properties, UUID}
  12.  
  13. class DorisStreamLoad2(props: PropertiesUtil) extends Serializable {
  14.   private val logger: Logger = LoggerFactory.getLogger(classOf[DorisStreamLoad2])
  15.  
  16.   private lazy val httpClientBuilder: HttpClientBuilder = HttpClients.custom.setRedirectStrategy(new DefaultRedirectStrategy() {
  17.     override protected def isRedirectable(method: String): Boolean = {
  18.       // If the connection target is FE, you need to deal with 307 redirect。
  19.       true
  20.     }
  21.   })
  22.  
  23.  
  24.   def loadJson(jsonData: String, mergeType: String, db: String, table: String): Unit = try {
  25.     val loadUrlPattern = "http://%s/api/%s/%s/_stream_load?"
  26.     val entity = new StringEntity(jsonData, "UTF-8")
  27.     val streamLoadProp = new Properties()
  28.     streamLoadProp.setProperty("merge_type", mergeType)
  29.     streamLoadProp.setProperty("format", "json")
  30.     streamLoadProp.setProperty("column_separator", ",")
  31.     streamLoadProp.setProperty("line_delimiter", ",")
  32.     streamLoadProp.setProperty("strip_outer_array", "true")
  33.     streamLoadProp.setProperty("exec_mem_limit", "6442450944")
  34.     streamLoadProp.setProperty("strict_mode", "true")
  35.     val httpClient = httpClientBuilder.build
  36.     val loadUrlStr = String.format(loadUrlPattern, props.doris_load_host, db, table)
  37.     try {
  38.       val builder = new HttpPutBuilder()
  39.       val label = UUID.randomUUID.toString
  40.       builder.setUrl(loadUrlStr)
  41.         .baseAuth(props.doris_user, props.doris_password)
  42.         .addCommonHeader()
  43.         .setLabel(label)
  44.         .setEntity(entity)
  45.         .addProperties(streamLoadProp)
  46.       handlePreCommitResponse(httpClient.execute(builder.build()))
  47.     }
  48.  
  49.     def handlePreCommitResponse(response: CloseableHttpResponse): Unit = {
  50.       val statusCode: Int = response.getStatusLine.getStatusCode
  51.       if (statusCode == 200 && response.getEntity != null) {
  52.         val loadResult: String = EntityUtils.toString(response.getEntity)
  53.         logger.info("load Result {}", loadResult)
  54.       } else {
  55.         throw new StreamLoadException("stream load error: " + response.getStatusLine.toString)
  56.       }
  57.  
  58.  
  59.     }
  60.  
  61.  
  62.   }
  63. }

4.SinkDoris.scala

  1. package com.zbkj.util
  2.  
  3. import com.alibaba.fastjson2.JSONArray
  4. import org.apache.flink.configuration.Configuration
  5. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
  6.  
  7. class SinkDoris(dorisStreamLoad:DorisStreamLoad2) extends RichSinkFunction[(String, String, String, JSONArray)]  {
  8.  
  9.  
  10.   override def open(parameters: Configuration): Unit = {}
  11.   /**
  12.    * 每个元素的插入都要调用一次invoke()方法进行插入操作
  13.    */
  14.   override def invoke(value:(String, String, String, JSONArray)): Unit = {
  15.     dorisStreamLoad.loadJson(value._4.toString,value._1,value._2,value._3)
  16.   }
  17.  
  18.   override def close(): Unit = {}
  19. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/398633
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号