devbox
凡人多烦事01
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

使用 Flink CDC 实现 MySQL 数据,表结构实时入 Apache Doris_flinkcdc同步mysql到doris

作者：凡人多烦事01 | 2024-05-22 05:20:38

赞

踩

flinkcdc同步mysql到doris

背景

  1. 现有数据库：mysql
  2. 数据：库表较多，每个企业用户一个分库，每个企业下的表均不同，无法做到聚合,且表可以被用户随意改动，增删改列等，增加表
  3. 分析：用户自定义分析，通过拖拽定义图卡，要求实时，点击确认即出现相应结果，其中有无法预判的过滤
  4. 问题：随业务增长，企业用户越来越多，mysql压力越来越大，已经出现一些图卡加载过慢[mysql sql]

同步流程

  1. 脚本读取mysql中需要同步的企业，在获取需要同步的表，以字段 member_id,table 字段存储doris中表A，
  2. 脚本读取doris 表A数据，获取mysql中的schema，通过转换，获取doris建表语句，连接doris执行语句
  3. cancel flink 任务，并重新启动flink任务【重启只适合添加新库，新表不用重启】
    1. 每次重启连接doris 表A，获取database，组装 databaseList,tableList，tablseList 使用正则，database1.*,database2.*,对库内所有表进行监听，这样可以达到mysql添加新表时将新表加入同步队列
    2. doris目前还不支持同步数据时同步修改表结构【据大佬说应该1.2+会支持】，不过cdc可以获取ddlsql，可以通过jdbc的方式连接doris去执行ddlsql,因为sql有点差异，需要转换才能执行，结合mysql新表，可以在ddl获取create 对doris进项建表
    3. 在将数据导入之doris时，速度导入过快都会出现导入失败，-235错误，可以使用控制读取binlog数量+window聚合 去批量导入
          如需要导入表B的数据有{"id":1,"name":"小明"},{"id":2,"name":"小红"}，如果执行两次put显然时不太合理的，可以使用jsonArr的方式[{"id":1,"name":"小明"},{"id":2,"name":"小红"}]一次导入

代码

        python 带码不在赘述，git：GitHub - xiaofeicn/MysqlToDorisTable

        Flink CDC

          flink中需要感知新表，每日重启时获取doris 表A数据，并组装成databaseList,tableList的参数，代码如下，代码有注释

        FlinkCDCMysql2Doris.scala

  1. package com.xxxx.mysql2doris
  2.  
  3.  
  4. import org.apache.flink.streaming.api.TimeCharacteristic
  5. import com.zbkj.util.{DorisStreamLoad, FlinkCDCSyncETL, KafkaUtil, PropertiesManager, PropertiesUtil, SinkDoris, SinkSchema}
  6. import com.ververica.cdc.connectors.mysql.source.MySqlSource
  7. import com.ververica.cdc.connectors.mysql.table.StartupOptions
  8. import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema
  9. import com.zbkj.util.KafkaUtil.proper
  10. import net.sf.json.JSONObject
  11. import org.apache.flink.api.common.eventtime.WatermarkStrategy
  12. import org.apache.flink.api.common.restartstrategy.RestartStrategies
  13. import org.apache.flink.api.common.time.Time
  14. import org.apache.flink.api.common.typeinfo.BasicTypeInfo
  15. import org.apache.flink.streaming.api.CheckpointingMode
  16. import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSource, DataStreamUtils}
  17. import org.apache.flink.streaming.api.environment.{CheckpointConfig, StreamExecutionEnvironment}
  18. import org.slf4j.{Logger, LoggerFactory}
  19. import org.apache.flink.api.java.io.jdbc.JDBCInputFormat
  20. import org.apache.flink.api.java.typeutils.RowTypeInfo
  21. import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
  22. import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
  23. import org.apache.kafka.connect.json.JsonConverterConfig
  24.  
  25. import java.util.Properties
  26. import java.util.concurrent.TimeUnit
  27. import scala.collection.JavaConverters.asScalaIteratorConverter
  28.  
  29. object FlinkCDCMysql2Doris {
  30.  
  31.   PropertiesManager.initUtil()
  32.   val props: PropertiesUtil = PropertiesManager.getUtil
  33.   val log: Logger = LoggerFactory.getLogger(this.getClass)
  34.  
  35.   def main(args: Array[String]): Unit = {
  36.     val env = StreamExecutionEnvironment.getExecutionEnvironment
  37.     // 并行度
  38.     env.setParallelism(props.parallelism)
  39.     env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
  40.  
  41.     /**
  42.      * checkpoint的相关设置
  43.      */
  44.     // 启用检查点，指定触发checkpoint的时间间隔（单位：毫秒，默认500毫秒），默认情况是不开启的
  45.     env.enableCheckpointing(10000L, CheckpointingMode.EXACTLY_ONCE)
  46.     // 设定Checkpoint超时时间，默认为10分钟
  47.     env.getCheckpointConfig.setCheckpointTimeout(600000)
  48.  
  49.     /** 设定两个Checkpoint之间的最小时间间隔，防止出现例如状态数据过大而导致Checkpoint执行时间过长，从而导致Checkpoint积压过多
  50.      * 最终Flink应用密切触发Checkpoint操作，会占用了大量计算资源而影响到整个应用的性能（单位：毫秒） */
  51.     env.getCheckpointConfig.setMinPauseBetweenCheckpoints(60000)
  52.     // 默认情况下，只有一个检查点可以运行
  53.     // 根据用户指定的数量可以同时触发多个Checkpoint，进而提升Checkpoint整体的效率
  54.     //    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
  55.     /** 外部检查点
  56.      * 不会在任务正常停止的过程中清理掉检查点数据，而是会一直保存在外部系统介质中，另外也可以通过从外部检查点中对任务进行恢复 */
  57.     env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
  58.  
  59.     /** 如果有更近的保存点时，是否将作业回退到该检查点 */
  60.     env.getCheckpointConfig.setPreferCheckpointForRecovery(true)
  61.     // 设置可以允许的checkpoint失败数
  62.     env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)
  63.     //设置可容忍的检查点失败数，默认值为0表示不允许容忍任何检查点失败
  64.     env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)
  65.  
  66.     /**
  67.      * 重启策略的配置
  68.      */
  69.     // 重启3次，每次失败后等待10000毫秒
  70.     //    env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.of(3, TimeUnit.MINUTES), Time.of(30, TimeUnit.SECONDS)))
  71.     env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L))
  72.  
  73.     /**
  74.      * 获取同步表配置
  75.      * database table
  76.      */
  77.     val inputMysql = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
  78.       .setDrivername("com.mysql.jdbc.Driver")
  79.       .setDBUrl("jdbc:mysql://%s:%d/%s".format(props.doris_host, props.doris_port, props.sync_config_db))
  80.       .setUsername(props.doris_user)
  81.       .setPassword(props.doris_password)
  82.       .setQuery("select member_id,sync_table from %s.%s".format(props.sync_config_db, props.sync_config_table))
  83.       .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
  84.       .finish()).uid("inputMysql")
  85.  
  86.  
  87.     val databaseName: DataStream[String] = inputMysql.map(line => line.getField(0).toString).uid("databaseName")
  88.     // 模糊监听
  89.     val tableName: DataStream[String] = inputMysql.map(line => line.getField(0).toString + ".*").uid("tableName")
  90.     val producer = KafkaUtil.getProducer
  91.  
  92.     val databaseIter = databaseName.executeAndCollect().asScala
  93.     val databaseList = databaseIter.toSet.mkString(",")
  94.  
  95.     val tableIter = tableName.executeAndCollect().asScala
  96.     val tableList = tableIter.toSet.mkString(",")
  97.     println("databaseList:", databaseList)
  98.     println("tableList:", tableList)
  99.     val customConverterConfigs = new java.util.HashMap[String, Object] {
  100.       put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric")
  101.     }
  102.     /**
  103.      *
  104.      * mysql source for doris
  105.      */
  106.     val mySqlSource = MySqlSource.builder[String]()
  107.       .hostname(props.rds_host)
  108.       .port(props.rds_port)
  109.       .databaseList(databaseList)
  110.       .tableList(tableList)
  111.       .username(props.rds_user)
  112.       .password(props.rds_password)
  113.       .serverId("11110")
  114.       .splitSize(props.split_size)
  115.       .fetchSize(props.fetch_size)
  116.       //       .startupOptions(StartupOptions.latest())
  117.       // 全量读取
  118.       .startupOptions(StartupOptions.initial())
  119.       .includeSchemaChanges(true)
  120.       // 发现新表，加入同步任务，需要在tableList中配置
  121.       .scanNewlyAddedTableEnabled(true)
  122.       .deserializer(new JsonDebeziumDeserializationSchema(false, customConverterConfigs)).build()
  123.  
  124.     val dataStreamSource: DataStreamSource[String] = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
  125.     val ddlSqlStream: DataStream[String] = dataStreamSource.filter(line => line.contains("historyRecord") && !line.contains("CHANGE COLUMN")).uid("ddlSqlStream")
  126.  
  127.     val dmlStream: DataStream[String] = dataStreamSource.filter(line => !line.contains("historyRecord") && !line.contains("CHANGE COLUMN")).uid("dmlStream")
  128.  
  129.     val ddlDataStream = FlinkCDCSyncETL.ddlFormat(ddlSqlStream)
  130.     val dmlDataStream = FlinkCDCSyncETL.binLogETL(dmlStream)
  131.     //    ddlDataStream.print()
  132.  
  133.     //producer 为了在数据同步后通知分析任务
  134.     val dorisStreamLoad = new DorisStreamLoad(props, producer)
  135.     ddlDataStream.addSink(new SinkSchema(props)).name("ALTER TABLE TO DORIS").uid("SinkSchema")
  136.     dmlDataStream.addSink(new SinkDoris(dorisStreamLoad)).name("Data TO DORIS").uid("SinkDoris")
  137.     env.execute("Flink CDC Mysql To Doris With Initial")
  138.  
  139.  
  140.   }
  141.  
  142.   case class dataLine(merge_type: String, db: String, table: String, data: String)
  143.  
  144.  
  145. }
FlinkCDCBinLogETL.scala
  1. package com.xxxx.util
  2.  
  3. import net.sf.json.JSONObject
  4. import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation}
  5. import org.apache.flink.api.java.tuple.Tuple4
  6. import org.apache.flink.streaming.api.datastream.DataStream
  7. import org.apache.flink.streaming.api.windowing.time.Time
  8.  
  9. import scala.collection.mutable.ArrayBuffer
  10. import scala.util.matching.Regex
  11.  
  12. object FlinkCDCSyncETL {
  13.  
  14.   def binLogETL(dataStreamSource: DataStream[String]): DataStream[org.apache.flink.api.java.tuple.Tuple4[String, String, String, String]] = {
  15.     /**
  16.      * 根据不同日志类型 匹配load doris方式
  17.      */
  18.     val tupleData: DataStream[org.apache.flink.api.java.tuple.Tuple4[String, String, String, String]] = dataStreamSource.map(line => {
  19.       var data: JSONObject = null
  20.       var mergetype = "APPEND"
  21.       val lineObj = JSONObject.fromObject(line)
  22.  
  23.       val source = lineObj.getJSONObject("source")
  24.       val db = source.getString("db")
  25.       val table = source.getString("table")
  26.       if ("d" == lineObj.getString("op")) {
  27.         val oo = lineObj.getJSONObject("before")
  28.         data = lineObj.getJSONObject("before")
  29.         mergetype = "DELETE"
  30.       } else if ("u" == lineObj.getString("op")) {
  31.         data = lineObj.getJSONObject("after")
  32.         mergetype = "MERGE"
  33.       } else if ("c" == lineObj.getString("op")) {
  34.         data = lineObj.getJSONObject("after")
  35.       } else if ("r" == lineObj.getString("op")) {
  36.         data = lineObj.getJSONObject("after")
  37.         mergetype = "APPEND"
  38.       }
  39.       new Tuple4[String, String, String, String](mergetype, db, table, data.toString)
  40.     }).returns(TypeInformation.of(new TypeHint[Tuple4[String, String, String, String]] {}))
  41.     tupleData
  42.     /**
  43.      * 窗口聚合数据，将相同load方式，db,table的json 数据组合为长字符串，
  44.      */
  45.     val byKeyData: DataStream[org.apache.flink.api.java.tuple.Tuple4[String, String, String, String]] = tupleData.keyBy(0, 1, 2)
  46.       .timeWindow(Time.milliseconds(1000))
  47.       .reduce((itemFirst, itemSecond) => new Tuple4(itemFirst.f0, itemFirst.f1, itemFirst.f2, itemFirst.f3 + "=-=-=" + itemSecond.f3))
  48.     byKeyData
  49.   }
  50.  
  51.   def ddlFormat(ddlDataStream: DataStream[String]): DataStream[String] = {
  52.  
  53.     val ddlStrDataStream: DataStream[String] = ddlDataStream.map(line => {
  54.       try {
  55.         val lineObj = JSONObject.fromObject(line)
  56.         val historyRecord = JSONObject.fromObject(lineObj.getString("historyRecord"))
  57.         val tableChangesArray = historyRecord.getJSONArray("tableChanges")
  58.         val tableChanges = JSONObject.fromObject(tableChangesArray.getJSONObject(0))
  59.         val tableChangeType = tableChanges.getString("type")
  60.         var ddlSql = ""
  61.  
  62.         val table = tableChanges.optJSONObject("table")
  63.         val primaryKeyColumnNames = table.getString("primaryKeyColumnNames").replace("[", "").replace("]", "").replace("\"", "")
  64.         val columnsArray = table.getJSONArray("columns")
  65.         // 建表转换
  66.         if (tableChangeType == "CREATE") {
  67.           val tableName = tableChanges.getString("id").replace("\"", "")
  68.           val columnsArrayBuffer = ArrayBuffer[String]()
  69.           columnsArray.forEach(line => {
  70.             val columnJson = JSONObject.fromObject(line)
  71.             val name = columnJson.getString("name")
  72.             val typeName = columnJson.getString("typeName")
  73.             val length = columnJson.optInt("length", 1)
  74.             val scale = columnJson.optInt("scale", 2)
  75.             val lastColumnType = matchColumnType(typeName, length, scale)
  76.             val lastColumn = s"$name $lastColumnType"
  77.             columnsArrayBuffer.+=(lastColumn)
  78.           })
  79.  
  80.           // 对列重新排序，主键依次放在最前面，避免错误Key columns should be a ordered prefix of the scheme
  81.           val keys = primaryKeyColumnNames.split(",")
  82.           for (indexOfCol <- 0 until keys.length) {
  83.             val col = keys(indexOfCol)
  84.             var columnFormat = ""
  85.             columnsArrayBuffer.foreach(column => {
  86.               if (column.startsWith(col)) {
  87.                 columnFormat = column
  88.               }
  89.  
  90.             })
  91.             val index = columnsArrayBuffer.indexOf(columnFormat)
  92.             columnsArrayBuffer.remove(index)
  93.             columnsArrayBuffer.insert(indexOfCol, columnFormat)
  94.  
  95.           }
  96.  
  97.  
  98.           val header = s"CREATE TABLE IF NOT EXISTS $tableName ("
  99.           val end = s""") UNIQUE KEY($primaryKeyColumnNames)  DISTRIBUTED BY HASH($primaryKeyColumnNames) BUCKETS 10  PROPERTIES ("replication_allocation" = "tag.location.default: 1")"""
  100.           ddlSql = header + columnsArrayBuffer.mkString(",") + end
  101.         } else if (tableChangeType == "ALTER") {
  102.           var ddl = historyRecord.getString("ddl").replace("\r\n", " ")
  103.           println(ddl)
  104.           if (ddl.startsWith("RENAME")) {
  105.             ddl = ddl.replace("`", "")
  106.             val arr = ddl.split("")
  107.             ddlSql = s"ALTER TABLE ${arr(2)} RENAME ${arr(4)}"
  108.           } else if (ddl.contains("DROP COLUMN")) {
  109.  
  110.             ddlSql = ddl
  111.           } else {
  112.             val dbTableName = tableChanges.getString("id").replace("\"", "")
  113.             val addColName = ddl.split(" ")(5).replace("`", "")
  114.             var colTpe = ""
  115.             columnsArray.forEach(line => {
  116.  
  117.               val columnJson = JSONObject.fromObject(line)
  118.  
  119.               if (columnJson.getString("name") == addColName) {
  120.                 val typeName = columnJson.getString("typeName")
  121.                 val length = columnJson.optInt("length", 1)
  122.                 val scale = columnJson.optInt("scale", 2)
  123.                 colTpe = matchColumnType(typeName, length, scale)
  124.               }
  125.  
  126.             })
  127.             if (ddl.contains("ADD COLUMN")) {
  128.  
  129.               ddlSql = s"ALTER TABLE $dbTableName ADD COLUMN $addColName $colTpe"
  130.             } else {
  131.  
  132.               ddlSql = s"ALTER TABLE $dbTableName MODIFY COLUMN $addColName $colTpe"
  133.             }
  134.  
  135.           }
  136.         }
  137.  
  138.         println(ddlSql)
  139.         ddlSql
  140.       }
  141.       catch {
  142.         case ex: Exception => println(ex)
  143.           "select 1"
  144.       }
  145.  
  146.     })
  147.     ddlStrDataStream
  148.   }
  149.  
  150.   def showCapital(x: Option[String]): String = x match {
  151.     case Some(s) => s
  152.     case None => "?"
  153.   }
  154.  
  155.   def matchColumnType(columnType: String, length: Int, scale: Int): String = {
  156.     var returnColumnType = "VARCHAR(255)"
  157.     columnType match {
  158.       case "INT UNSIGNED" => returnColumnType = s"INT($length)"
  159.       case "INT" => returnColumnType = s"INT($length)"
  160.       case "TINYINT" => returnColumnType = s"TINYINT($length)"
  161.       case "VARCHAR" => returnColumnType = s"VARCHAR(${length * 3})"
  162.       case "BIGINT" => returnColumnType = s"BIGINT(${length})"
  163.       case "TINYTEXT" => returnColumnType = s"TINYTEXT"
  164.       case "LONGTEXT" => returnColumnType = s"STRING"
  165.       case "TEXT" => returnColumnType = s"STRING"
  166.       case "DECIMAL" => returnColumnType = s"DECIMAL($length,$scale)"
  167.       case "VARBINARY" => returnColumnType = s"STRING"
  168.       case "TIMESTAMP" => returnColumnType = s"STRING"
  169.       case "ENUM" => returnColumnType = s"TINYINT"
  170.       case "MEDIUMINT" => returnColumnType = s"INT"
  171.       case "SMALLINT" => returnColumnType = s"SMALLINT"
  172.       case "MEDIUMTEXT" => returnColumnType = s"STRING"
  173.       case _ => returnColumnType = s"STRING"
  174.     }
  175.     returnColumnType
  176.   }
  177.  
  178. }
DorisStreamLoad.scala
  1. package com.xxxx.util
  2.  
  3. import net.sf.json.JSONObject
  4. import net.sf.json.JSONArray
  5. import org.apache.http.HttpHeaders
  6. import org.apache.http.client.methods.HttpPut
  7. import org.apache.http.entity.StringEntity
  8. import org.apache.http.entity.BufferedHttpEntity
  9. import org.apache.http.impl.client.{DefaultRedirectStrategy, HttpClientBuilder, HttpClients}
  10. import org.apache.http.util.EntityUtils
  11. import org.slf4j.{Logger, LoggerFactory}
  12. import org.apache.commons.codec.binary.Base64
  13. import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
  14.  
  15. import java.io.IOException
  16. import java.nio.charset.StandardCharsets
  17. import java.util.UUID
  18.  
  19. class DorisStreamLoad(props: PropertiesUtil,producer:KafkaProducer[String, String]) extends Serializable {
  20.  
  21.  
  22.   lazy val httpClientBuilder: HttpClientBuilder = HttpClients.custom.setRedirectStrategy(new DefaultRedirectStrategy() {
  23.     override protected def isRedirectable(method: String): Boolean = {
  24.       // If the connection target is FE, you need to deal with 307 redirect。
  25.       true
  26.     }
  27.   })
  28.  
  29.  
  30.   def loadJson(jsonData: String, mergeType: String, db: String, table: String): Unit = try {
  31.     val loadUrlPattern = "http://%s/api/%s/%s/_stream_load?"
  32.     val arr = jsonData.split("=-=-=")
  33.     val jsonArray = new JSONArray()
  34.     for (line <- arr) {
  35.       try {
  36.         val js = JSONObject.fromObject(line)
  37.         jsonArray.add(js)
  38.       } catch {
  39.         case e: Exception =>
  40.           println(e)
  41.           println(line)
  42.       }
  43.  
  44.     }
  45.     val jsonArrayStr = jsonArray.toString()
  46.     val client = httpClientBuilder.build
  47.     val loadUrlStr = String.format(loadUrlPattern, props.doris_load_host, db, table)
  48.     try {
  49.       val put = new HttpPut(loadUrlStr)
  50.       put.removeHeaders(HttpHeaders.CONTENT_LENGTH)
  51.       put.removeHeaders(HttpHeaders.TRANSFER_ENCODING)
  52.       put.setHeader(HttpHeaders.EXPECT, "100-continue")
  53.       put.setHeader(HttpHeaders.AUTHORIZATION, basicAuthHeader)
  54.       val label = UUID.randomUUID.toString
  55.       // You can set stream load related properties in the Header, here we set label and column_separator.
  56.       put.setHeader("label", label)
  57.       put.setHeader("merge_type", mergeType)
  58.       //      put.setHeader("two_phase_commit", "true")
  59.       put.setHeader("column_separator", ",")
  60.       put.setHeader("format", "json")
  61.       put.setHeader("strip_outer_array", "true")
  62.       put.setHeader("exec_mem_limit", "6442450944")
  63.       val entity = new StringEntity(jsonArrayStr, "UTF-8")
  64.  
  65.       put.setEntity(entity)
  66.  
  67.       try {
  68.         val response = client.execute(put)
  69.         try {
  70.           var loadResult = ""
  71.           if (response.getEntity != null) {
  72.             loadResult = EntityUtils.toString(response.getEntity)
  73.           }
  74.           val statusCode = response.getStatusLine.getStatusCode
  75.           if (statusCode != 200) {
  76.             throw new IOException("Stream load failed. status: %s load result: %s".format(statusCode, loadResult))
  77.           }
  78.  
  79.         } finally if (response != null) {
  80.           response.close()
  81.         }
  82.       }
  83.     }
  84.     finally
  85.       if (client != null) client.close()
  86.   }
  87.  
  88.   /**
  89.    * Construct authentication information, the authentication method used by doris here is Basic Auth
  90.    *
  91.    */
  92.   def basicAuthHeader: String = {
  93.     val tobeEncode = props.doris_user + ":" + props.doris_password
  94.     val encoded = Base64.encodeBase64(tobeEncode.getBytes(StandardCharsets.UTF_8))
  95.     "Basic " + new String(encoded)
  96.   }
  97.  
  98.  
  99. }
SinkDoris.scala
  1. package com.xxxx.util
  2.  
  3.  
  4. import org.apache.flink.api.java.tuple.Tuple4
  5. import com.zbkj.mysql2doris.FlinkCDCMysql2Doris.dataLine
  6. import net.sf.json.JSONObject
  7. import org.apache.flink.streaming.api.functions.sink.SinkFunction
  8.  
  9.  
  10. class SinkDoris(dorisStreamLoad:DorisStreamLoad) extends SinkFunction[Tuple4[String, String, String, String]]  {
  11.  
  12. //  val  dorisStreamLoad:DorisStreamLoadT=null
  13.   /**
  14.    * 在open()方法中建立连接，这样不用每次invoke的时候都要建立连接和释放连接。
  15.    */
  16. //   def open(parameters: Configuration): Unit = {
  17. //     super
  18. //     super.open(parameters);
  19. //
  20. //  }
  21.  
  22.   /**
  23.    * 每个元素的插入都要调用一次invoke()方法进行插入操作
  24.    */
  25.   override def invoke(value:Tuple4[String, String, String, String]): Unit = {
  26.  
  27.     dorisStreamLoad.loadJson(value.f3,value.f0,value.f1,value.f2)
  28.     val producer = KafkaUtil.getProducer
  29.     val json = new JSONObject()
  30.     json.put("db",value.f2)
  31.     json.put("table",value.f3)
  32.     KafkaUtil.sendKafkaMsg(producer, json.toString, "sync_table")
  33.  
  34.   }
  35.  
  36. }
SinkSchema.scala
  1. package com.xxxx.util
  2.  
  3. import org.apache.flink.configuration.Configuration
  4. import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
  5.  
  6. import java.sql.{Connection, DriverManager, PreparedStatement}
  7.  
  8. class SinkSchema(props:PropertiesUtil) extends RichSinkFunction[String]  {
  9.   var conn: Connection = _
  10.   var ps : PreparedStatement  = _
  11.   var mysqlPool: MysqlPool = _
  12.  
  13.   override def open(parameters: Configuration): Unit = {
  14.     super.open(parameters)
  15.     mysqlPool = MysqlManager.getMysqlPool
  16.     conn = mysqlPool.getConnection
  17.        conn.setAutoCommit(false)
  18.   }
  19.  
  20.   override def close(): Unit = {
  21.     super.close()
  22.     if (conn != null) {
  23.       conn.close()
  24.     }
  25.     if (ps != null) {
  26.       ps.close()
  27.     }
  28.   }
  29.  
  30.   override def invoke(sql: String, context: SinkFunction.Context): Unit = {
  31.     super.invoke(sql, context)
  32.     if (sql !="" && sql.nonEmpty){
  33.       ps = conn.prepareStatement(sql)
  34.       try {
  35.         ps.execute()
  36.       }catch {
  37.         case ex:Exception=>println(ex)
  38.       }
  39.  
  40.       conn.commit()
  41.  
  42.     }
  43. //    conn.close()
  44.   }
  45.  
  46. }
PropertiesUtil.scala
  1. package com.xxxx.util
  2.  
  3. import java.io.FileInputStream
  4. import java.util.Properties
  5.  
  6. /**
  7.  * propertiesUtil
  8.  *
  9.  */
  10. class PropertiesUtil extends Serializable {
  11.  
  12.  
  13.   private val props = new Properties()
  14.  
  15.   var doris_host = ""
  16.   var doris_port = 0
  17.   var doris_user = ""
  18.   var doris_password = ""
  19.  
  20.   var database_list = ""
  21.   var table_list = ""
  22.  
  23.   var mysql_host = ""
  24.   var mysql_port = 0
  25.   var mysql_user = ""
  26.   var mysql_password = ""
  27.   var doris_load_host = ""
  28.  
  29.   var rds_host = ""
  30.   var rds_port = 0
  31.   var rds_user = ""
  32.   var rds_password = ""
  33.   var rds_database = ""
  34.  
  35. //  var sync_database_select_sql = ""
  36. //  var sync_table_select_sql = ""
  37. //  var sync_config_host = ""
  38. //  var sync_config_port = 0
  39. //  var sync_config_user = ""
  40. //  var sync_config_password = ""
  41.   var sync_config_db = ""
  42.   var sync_config_table = ""
  43.   var sync_redis_table = ""
  44.   var address_table = ""
  45.  
  46.   var parallelism = 0
  47.   var split_size = 0
  48.   var fetch_size = 0
  49.  
  50.   var bootstrap_servers = ""
  51.   var topic = ""
  52.   var group_id = ""
  53.   var offset_mode = ""
  54.  
  55.   // reids
  56.   var redis_max_total: Int = 0
  57.   var redis_max_idle: Int = 0
  58.   var redis_min_idle: Int = 0
  59.   var redis_host = ""
  60.   var redis_port: Int = 0
  61.   var redis_timeout: Int = 0
  62.   var redis_password = ""
  63.   var redis_db_index: Int = 0
  64.   var prefix = "0"
  65.  
  66.  
  67.   def init(filePath: String): Unit = {
  68.     props.load(new FileInputStream(filePath))
  69.  
  70.     // hdfs
  71.     doris_host = props.getProperty("doris.host")
  72.     doris_port = props.getProperty("doris.port").toInt
  73.     doris_user = props.getProperty("doris.user")
  74.     doris_password = props.getProperty("doris.password")
  75.  
  76.     database_list = props.getProperty("database.list")
  77.     table_list = props.getProperty("table.list")
  78.  
  79.     mysql_host = props.getProperty("mysql.host")
  80.     mysql_port = props.getProperty("mysql.port").toInt
  81.     mysql_user = props.getProperty("mysql.user")
  82.     mysql_password = props.getProperty("mysql.password")
  83.     doris_load_host = props.getProperty("doris.load.host")
  84.  
  85.  
  86.     rds_host = props.getProperty("rds.host")
  87.     rds_port = props.getProperty("rds.port").toInt
  88.     rds_user = props.getProperty("rds.user")
  89.     rds_password = props.getProperty("rds.password")
  90.     rds_database = props.getProperty("rds.database")
  91.     sync_config_db = props.getProperty("sync.config.db")
  92.     sync_config_table = props.getProperty("sync.config.table")
  93.     sync_redis_table = props.getProperty("sync.redis.table")
  94.     address_table = props.getProperty("address.table")
  95.  
  96.  
  97.     parallelism = props.getProperty("parallelism").toInt
  98.     split_size = props.getProperty("split.size").toInt
  99.     fetch_size = props.getProperty("fetch.size").toInt
  100.  
  101.     bootstrap_servers = props.getProperty("bootstrap.servers")
  102.     topic = props.getProperty("topic")
  103.     group_id = props.getProperty("group.id")
  104.     offset_mode = props.getProperty("offset.mode")
  105.  
  106.  
  107.     // reids
  108.     redis_max_total = props.getProperty("redis.max.total").toInt
  109.     redis_max_idle = props.getProperty("redis.max.idle").toInt
  110.     redis_min_idle = props.getProperty("redis.min.idle").toInt
  111.     redis_host = props.getProperty("redis.redis.host")
  112.     redis_port = props.getProperty("redis.redis.port").toInt
  113.     redis_timeout = props.getProperty("redis.redis.timeout").toInt
  114.     redis_password = props.getProperty("redis.password")
  115.     redis_db_index = props.getProperty("redis.db.index").toInt
  116.  
  117.     prefix = props.getProperty("redis.key.prefix")
  118.  
  119.  
  120.   }
  121.  
  122.   def stringToInt(prop: String): Int = {
  123.     try {
  124.       prop.toInt
  125.     } catch {
  126.       case ex: Exception => {
  127.         0
  128.       }
  129.     }
  130.   }
  131. }
  132.  
  133. //惰性单例，真正计算时才初始化对象
  134. object PropertiesManager {
  135.   @volatile private var propertiesUtil: PropertiesUtil = _
  136.  
  137.   def getUtil: PropertiesUtil = {
  138.     propertiesUtil
  139.   }
  140.  
  141.   def initUtil(): Unit = {
  142.     var filePath = "config.properties"
  143. //        filePath = this.getClass.getResource("/").toString.replace("file:", "") + "config.properties"
  144.     filePath = "/opt/flink-1.13.6/job/mysql2doris/config.properties"
  145.     if (propertiesUtil == null) {
  146.       propertiesUtil = new PropertiesUtil
  147.     }
  148.     propertiesUtil.init(filePath)
  149.     //    propertiesUtil.evn = evn
  150.   }
  151. }

若有疑问请留言或者 加入857技术社区

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

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

  

闽ICP备14008679号