Flink CDC2，20天内看完这套GitHub标星18k+的大数据开发资料_flink cdc github

println(dataBaseList, tableList)
val debeziumProps = new Properties()
debeziumProps.setProperty(“debezium.snapshot.mode”,“never”)
val mysqlSource = MySqlSource.builderString
.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

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数大数据工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年大数据全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）

的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上大数据开发知识点，真正体系化！**

由于文件比较大，这里只是将部分目录大纲截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且后续会持续更新

如果你觉得这些内容对你有帮助，可以添加VX：vip204888 （备注大数据获取）
[外链图片转存中…(img-HnhuWNzt-1712517044421)]