一、核心類功能

• MySqlSourceEnumerator: 在 JobManager 上以單并行度運(yùn)行, 會(huì)生成數(shù)據(jù)分片并將它們分配給 SourceReader
• MySqlSourceReader: 在 TaskManagers 上的 SourceOperators 并行運(yùn)行, 會(huì)請(qǐng)求MySqlSourceEnumerator進(jìn)行分片并進(jìn)行處理，例如讀取分片所表示的文件或日志分區(qū)

• MySqlSplit(分片): 是對(duì)一部分 source 數(shù)據(jù)的包裝，如一個(gè)文件或者日志分區(qū)。分片是 source 進(jìn)行任務(wù)分配和數(shù)據(jù)并行讀取的基本粒度

  
  
  a6591e1ac5074d668c9a562314015f40~noop.png

二、CDC同步示例

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.catalog.ObjectPath;

import java.time.Duration;
import java.util.Properties;

public class MysqlCDC {
    public static void main(String[] args) throws Exception {
        Properties debeziumProperties = new Properties();
        debeziumProperties.put("decimal.handling.mode", "String");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("master1")
                .port(3306)
                .scanNewlyAddedTableEnabled(true) // 開啟支持新增表
                .databaseList("test") // set captured database
                .tableList("test.orders") // set captured table
                .username("root")
                .password("123456")
                .serverTimeZone("UTC")
                .serverId("20-40")
                .includeSchemaChanges(true)
                .heartbeatInterval(Duration.ofSeconds(30))
                .startupOptions(StartupOptions.initial())
                .scanNewlyAddedTableEnabled(true)
                .chunkKeyColumn(new ObjectPath("test","orders"),"order_id")
                .debeziumProperties(debeziumProperties)
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(
                configuration
        );
        // enable checkpoint
        env.enableCheckpointing(3000);
        // 設(shè)置本地
        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-ck");
        env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                // set 4 parallel source tasks
                .setParallelism(4)
                .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
        env.execute("Print MySQL Snapshot + Binlog");
    }
}

主要是構(gòu)建MySqlSource對(duì)象,啟動(dòng)flink任務(wù), MySqlSource繼承了Source, 在任務(wù)啟動(dòng)時(shí)會(huì)在JobManager節(jié)點(diǎn)上執(zhí)行createEnumerator方法創(chuàng)建MySqlSourceEnumerator對(duì)象, 并執(zhí)行其start方法, 在TaskManager節(jié)點(diǎn)上會(huì)執(zhí)行createReader方法創(chuàng)建MySqlSourceReader對(duì)象,并執(zhí)行其start方法

三、MySqlSourceEnumerator

MySqlSourceEnumerator對(duì)象創(chuàng)建

 @Override
    public SplitEnumerator<MySqlSplit, PendingSplitsState> createEnumerator(
            SplitEnumeratorContext<MySqlSplit> enumContext) {
        MySqlSourceConfig sourceConfig = configFactory.createConfig(0, ENUMERATOR_SERVER_NAME);

        /**
         *  1.校驗(yàn) mysql 版本必須>=5.6 以上
         *  2.校驗(yàn)mysql format必須時(shí)ROW模式
         *  3.校驗(yàn)mysql binlog_row_image 必須時(shí)FULL
         *  4.校驗(yàn)time zone時(shí)區(qū)參數(shù)是否和mysql設(shè)置的一致
         * */
        final MySqlValidator validator = new MySqlValidator(sourceConfig);
        validator.validate();

        final MySqlSplitAssigner splitAssigner;
        /**
         *  如果CDC啟動(dòng)參數(shù)startupOptions設(shè)置的是INITIAL, 代表, 先同步全量數(shù)據(jù), 再同步增量數(shù)據(jù)
         *  StartupMode 屬性有以下幾種取值：
         *      1、StartupMode.LATEST_OFFSET：Binlog增量, 表示任務(wù)在啟動(dòng)時(shí)從最新的偏移量開始讀取數(shù)據(jù)。
         *      2、StartupMode.EARLIEST_OFFSET：Binlog增量,表示任務(wù)在啟動(dòng)時(shí)從最早的偏移量開始讀取數(shù)據(jù)。
         *      3、StartupMode.SPECIFIC_OFFSET：Binlog增量, 表示任務(wù)在啟動(dòng)時(shí)從指定的偏移量開始讀取數(shù)據(jù)。
         *      4、StartupMode.TIMESTAMP：表示任務(wù)在啟動(dòng)時(shí)從指定的時(shí)間戳開始讀取數(shù)據(jù)。
         */
        if (sourceConfig.getStartupOptions().startupMode == StartupMode.INITIAL) {
            try (JdbcConnection jdbc = openJdbcConnection(sourceConfig)) {
                boolean isTableIdCaseSensitive = DebeziumUtils.isTableIdCaseSensitive(jdbc);
                // MySqlHybridSplitAssigner 混合方式, 實(shí)現(xiàn)先全量, 再增量
                splitAssigner =
                        new MySqlHybridSplitAssigner(
                                sourceConfig,
                                enumContext.currentParallelism(),
                                new ArrayList<>(),
                                isTableIdCaseSensitive);
            } catch (Exception e) {
                throw new FlinkRuntimeException(
                        "Failed to discover captured tables for enumerator", e);
            }
        } else {
            // MySqlBinlogSplitAssigner Binlog增量
            splitAssigner = new MySqlBinlogSplitAssigner(sourceConfig);
        }

        return new MySqlSourceEnumerator(enumContext, sourceConfig, splitAssigner);
    }

四、MySqlSourceReader

 @Override
    public SourceReader<T, MySqlSplit> createReader(SourceReaderContext readerContext)
            throws Exception {
        // create source config for the given subtask (e.g. unique server id)
        MySqlSourceConfig sourceConfig =
                configFactory.createConfig(readerContext.getIndexOfSubtask());
        //數(shù)據(jù)存儲(chǔ)隊(duì)列
        FutureCompletingBlockingQueue<RecordsWithSplitIds<SourceRecords>> elementsQueue =
                new FutureCompletingBlockingQueue<>();

        final Method metricGroupMethod = readerContext.getClass().getMethod("metricGroup");
        metricGroupMethod.setAccessible(true);
        final MetricGroup metricGroup = (MetricGroup) metricGroupMethod.invoke(readerContext);

        // reader metrics 相關(guān)指標(biāo)
        /**
         *  其中currentEmitEventTimeLag 指標(biāo)記錄的是 Source 發(fā)送一條記錄到下游節(jié)點(diǎn)的時(shí)間點(diǎn)和該記錄在 DB 里產(chǎn)生時(shí)間點(diǎn)差值，
         *  用于衡量數(shù)據(jù)從 DB 產(chǎn)生到離開 Source 節(jié)點(diǎn)的延遲。用戶可以通過該指標(biāo)判斷 source 是否進(jìn)入了 binlog 讀取階段
         *
         *  如果等于0表示處于全量階段, 非0表示處于增量階段
         * */
        final MySqlSourceReaderMetrics sourceReaderMetrics =
                new MySqlSourceReaderMetrics(metricGroup);
        sourceReaderMetrics.registerMetrics();
        MySqlSourceReaderContext mySqlSourceReaderContext =
                new MySqlSourceReaderContext(readerContext);
        // 通過supplier函數(shù)構(gòu)建一個(gè)SplitReader
        Supplier<MySqlSplitReader> splitReaderSupplier =
                () ->
                        // 拿到每個(gè)reader的config和對(duì)應(yīng)的subtask index
                        new MySqlSplitReader(
                                sourceConfig,
                                readerContext.getIndexOfSubtask(),
                                mySqlSourceReaderContext);
        // 構(gòu)建了一個(gè)具體的sourceReader
        return new MySqlSourceReader<>(
                elementsQueue,
                splitReaderSupplier,
                new MySqlRecordEmitter<>(
                        deserializationSchema,
                        sourceReaderMetrics,
                        sourceConfig.isIncludeSchemaChanges()),
                readerContext.getConfiguration(),
                mySqlSourceReaderContext,
                sourceConfig);
    }