环境
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 避免聚合导致背压，字符串在数据量较大时拼接效率太低文章来源：https://www.toymoban.com/news/detail-510972.html
Flink cdc 代码
1、FlinkSingleSync.scala
文章来源地址https://www.toymoban.com/news/detail-510972.html
package com.zbkj.sync
import com.alibaba.fastjson2.{JSON, JSONObject,JSONArray}
import com.ververica.cdc.connectors.mysql.source.MySqlSource
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.json.JsonConverterConfig
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema
import com.zbkj.util.SinkBuilder.getKafkaSink
import com.zbkj.util._
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._
import java.util.Properties
object FlinkSingleSync {
PropertiesManager.initUtil()
val props: PropertiesUtil = PropertiesManager.getUtil
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 并行度
env.setParallelism(props.parallelism)
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
val parameters: ParameterTool = ParameterTool.fromArgs(args)
val memberID = parameters.getInt("memberID", 0)
val source = parameters.get("source", "")
val log = parameters.getBoolean("log", true)
if (memberID == 0) {
sys.exit(0)
}
val thisMember = "ttk_member_%d".format(memberID)
val jobName = "Sync Member %d".format(memberID)
val syncTopic = "sync_data_%d".format(memberID)
println(syncTopic)
val sourceFormat = SourceFormat.sourceFormat(source)
env.setParallelism(4)
/**
* checkpoint的相关设置 */
// 启用检查点，指定触发checkpoint的时间间隔（单位：毫秒，默认500毫秒），默认情况是不开启的
env.enableCheckpointing(1000L, CheckpointingMode.EXACTLY_ONCE)
// 设定Checkpoint超时时间，默认为10分钟
env.getCheckpointConfig.setCheckpointTimeout(600000)
/**
* 设置检查点路径 */
env.getCheckpointConfig.setCheckpointStorage("file:///data/flink-checkpoints/sync/%d".format(memberID))
/** 设定两个Checkpoint之间的最小时间间隔，防止出现例如状态数据过大而导致Checkpoint执行时间过长，从而导致Checkpoint积压过多
* 最终Flink应用密切触发Checkpoint操作，会占用了大量计算资源而影响到整个应用的性能（单位：毫秒） */
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(60000)
// 默认情况下，只有一个检查点可以运行
// 根据用户指定的数量可以同时触发多个Checkpoint，进而提升Checkpoint整体的效率
//env.getCheckpointConfig.setMaxConcurrentCheckpoints(2)
/** 外部检查点
* 不会在任务正常停止的过程中清理掉检查点数据，而是会一直保存在外部系统介质中，另外也可以通过从外部检查点中对任务进行恢复 */
env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
// env.getCheckpointConfig.setPreferCheckpointForRecovery(true)
// 设置可以允许的checkpoint失败数
env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3)
//设置可容忍的检查点失败数，默认值为0表示不允许容忍任何检查点失败
env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)
env.disableOperatorChaining()
/**
* 重启策略的配置
* 重启3次，每次失败后等待10000毫秒
*/
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 30000L))
val dataBaseList = thisMember
var tableList = thisMember + ".*"
if (!log) {
tableList = "lb_crm_customer_log|.*(?<!_log)$"
}
val dorisStreamLoad = new DorisStreamLoad2(props)
// numeric 类型转换
val customConverterConfigs = new java.util.HashMap[String, Object] {
put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, "numeric")
}
/**
* mysql source for doris */
println(dataBaseList, tableList)
val debeziumProps = new Properties()
debeziumProps.setProperty("debezium.snapshot.mode","never")
val mysqlSource = MySqlSource.builder[String]()
.hostname(sourceFormat.getString("sourceHost"))
.port(sourceFormat.getIntValue("sourcePort"))
.databaseList(dataBaseList)
//^((?!lb_admin_log|lb_bugs).)*$
// lb_admin_log、lb_bugs为不需要同步表
.tableList(props.regular_expression)
.username(sourceFormat.getString("sourceUsername"))
.password(sourceFormat.getString("sourcePassword"))
.debeziumProperties(debeziumProps)
// 全量读取
.startupOptions(StartupOptions.initial())
.includeSchemaChanges(true)
// 发现新表，加入同步任务，需要在tableList中配置
.scanNewlyAddedTableEnabled(true)
.deserializer(new JsonDebeziumDeserializationSchema(false, customConverterConfigs)).build()
val streamSource: DataStream[JSONObject] = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
.map(line => JSON.parseObject(line)).setParallelism(4)
val DDLSqlStream: DataStream[JSONObject] = streamSource.filter(line => !line.containsKey("op")).uid("ddlSqlStream")
val DMLStream: DataStream[JSONObject] = streamSource.filter(line => line.containsKey("op")).uid("dmlStream")
/**
* 首次全量同步时 时间窗口内几乎为一个表数据，此时下面操作会数据倾斜
* 在binLogETLOne 中对表加随机数后缀 使其均匀分布
* 聚合操作之后再将tableName转换为实际表
*/
val DMLDataStream = FlinkCDCSyncETL.binLogETLOne(DMLStream)
val keyByDMLDataStream:DataStream[(String, String, String, JSONArray)] = DMLDataStream.keyBy(keys => (keys._1, keys._2, keys._3))
.timeWindow(Time.milliseconds(props.window_time_milliseconds))
.reduce((itemFirst, itemSecond) => (itemFirst._1, itemFirst._2, itemFirst._3,combineJsonArray(itemFirst._4,itemSecond._4)))
.map(line=>(line._1,line._2,line._3.split("-")(0),line._4))
.name("分组聚合").uid("keyBy")
keyByDMLDataStream.addSink(new SinkDoris(dorisStreamLoad)).name("数据写入Doris").uid("SinkDoris").setParallelism(4)
val DDLKafkaSink=getKafkaSink("schema_change")
DDLSqlStream.map(jsObj => jsObj.toJSONString()).sinkTo(DDLKafkaSink).name("同步DDL入Kafka").uid("SinkDDLKafka")
val kafkaSink=getKafkaSink(syncTopic)
keyByDMLDataStream.map(line=>(line._2,line._3,1)).filter(!_._2.endsWith("_sql"))
.keyBy(keys => (keys._1, keys._2))
.window(TumblingProcessingTimeWindows.of(Time.seconds(1))).sum(2)
.map(line =>{
val json = new JSONObject()
json.put("member_id", line._1)
json.put("table", line._2)
json.toJSONString()
}).sinkTo(kafkaSink).name("同步数据库表入Kafka").uid("syncDataTableToKafka")
env.execute(jobName)
}
def combineJsonArray(jsr1:JSONArray,jsr2:JSONArray): JSONArray ={
jsr1.addAll(jsr2)
jsr1
}
}2.FlinkCDCSyncETL.scala
package com.zbkj.util
import com.alibaba.fastjson2.{JSON, JSONArray, JSONObject}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.DataStream
import java.util.Random
object FlinkCDCSyncETL {
def binLogETLOne(dataStreamSource: DataStream[JSONObject]): DataStream[(String, String, String, JSONArray)] = {
/**
* 根据不同日志类型 匹配load doris方式
*/
val tupleData: DataStream[(String, String, String, JSONArray)] = dataStreamSource.map(line => {
var data: JSONObject = new JSONObject()
var jsr: JSONArray = new JSONArray()
var mergeType = "APPEND"
val source = line.getJSONObject("source")
val db = source.getString("db")
val table = source.getString("table")
val op=line.getString("op")
if ("d" == op) {
data = line.getJSONObject("before")
mergeType = "DELETE"
} else if ("u" == op) {
data = line.getJSONObject("after")
mergeType = "APPEND"
} else if ("c" == op) {
data = line.getJSONObject("after")
} else if ("r" == op) {
data = line.getJSONObject("after")
mergeType = "APPEND"
}
jsr.add(data)
Tuple4(mergeType, db, table+ "-" + new Random().nextInt(4), jsr)
})
tupleData
}
}
3.DorisStreamLoad2.scala
package com.zbkj.util
import org.apache.doris.flink.exception.StreamLoadException
import org.apache.doris.flink.sink.HttpPutBuilder
import org.apache.http.client.methods.CloseableHttpResponse
import org.apache.http.entity.StringEntity
import org.apache.http.impl.client.{DefaultRedirectStrategy, HttpClientBuilder, HttpClients}
import org.apache.http.util.EntityUtils
import org.slf4j.{Logger, LoggerFactory}
import java.util.{Properties, UUID}
class DorisStreamLoad2(props: PropertiesUtil) extends Serializable {
private val logger: Logger = LoggerFactory.getLogger(classOf[DorisStreamLoad2])
private lazy val httpClientBuilder: HttpClientBuilder = HttpClients.custom.setRedirectStrategy(new DefaultRedirectStrategy() {
override protected def isRedirectable(method: String): Boolean = {
// If the connection target is FE, you need to deal with 307 redirect。
true
}
})
def loadJson(jsonData: String, mergeType: String, db: String, table: String): Unit = try {
val loadUrlPattern = "http://%s/api/%s/%s/_stream_load?"
val entity = new StringEntity(jsonData, "UTF-8")
val streamLoadProp = new Properties()
streamLoadProp.setProperty("merge_type", mergeType)
streamLoadProp.setProperty("format", "json")
streamLoadProp.setProperty("column_separator", ",")
streamLoadProp.setProperty("line_delimiter", ",")
streamLoadProp.setProperty("strip_outer_array", "true")
streamLoadProp.setProperty("exec_mem_limit", "6442450944")
streamLoadProp.setProperty("strict_mode", "true")
val httpClient = httpClientBuilder.build
val loadUrlStr = String.format(loadUrlPattern, props.doris_load_host, db, table)
try {
val builder = new HttpPutBuilder()
val label = UUID.randomUUID.toString
builder.setUrl(loadUrlStr)
.baseAuth(props.doris_user, props.doris_password)
.addCommonHeader()
.setLabel(label)
.setEntity(entity)
.addProperties(streamLoadProp)
handlePreCommitResponse(httpClient.execute(builder.build()))
}
def handlePreCommitResponse(response: CloseableHttpResponse): Unit = {
val statusCode: Int = response.getStatusLine.getStatusCode
if (statusCode == 200 && response.getEntity != null) {
val loadResult: String = EntityUtils.toString(response.getEntity)
logger.info("load Result {}", loadResult)
} else {
throw new StreamLoadException("stream load error: " + response.getStatusLine.toString)
}
}
}
}4.SinkDoris.scala
package com.zbkj.util
import com.alibaba.fastjson2.JSONArray
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
class SinkDoris(dorisStreamLoad:DorisStreamLoad2) extends RichSinkFunction[(String, String, String, JSONArray)] {
override def open(parameters: Configuration): Unit = {}
/**
* 每个元素的插入都要调用一次invoke()方法进行插入操作
*/
override def invoke(value:(String, String, String, JSONArray)): Unit = {
dorisStreamLoad.loadJson(value._4.toString,value._1,value._2,value._3)
}
override def close(): Unit = {}
}
到了这里，关于Flink CDC2.4 整库实时同步MySql 到Doris的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
