基於chunjun純鈞的增量資料同步

目前我司的巨量資料平臺使用的是flink技術棧，底層的聯結器外掛使用的是國產的chunjun外掛，在使用chunjun的過程中也遇到了很多問題，本次記錄下在SQL模式的情況下怎麼支援增量的資料同步

chunjun的官網檔案對增量同步已經做出了一定的說明

純鈞官方
根據檔案我編寫了一個SQL指令碼

create table `source` (
        `sfzh` STRING COMMENT '',
        `xm` STRING COMMENT '',
        `xb` STRING COMMENT '',
        `xbdm` STRING COMMENT '',
        `jzdz` STRING COMMENT '',
        `fzrq` DATE COMMENT '',
        `dsc_biz_record_id` STRING COMMENT ''
) with (
        'connector' = 'mysql-x',
        'url' = 'jdbc:mysql://:/?useSSL=false&useInformationSchema=true&nullCatalogMeansCurrent=true',
        'table-name' = '',
        'username' = '',
        'password' = '',
        'scan.fetch-size' = '1024',
        'scan.increment.column' = 'fzrq',
        --'scan.increment.column-type' = 'date',
        'scan.start-location' = '1659974400000'
);

create table `sink` (
        `sfzh` STRING COMMENT '',
        `xm` STRING COMMENT '',
        `xb` STRING COMMENT '',
        `xbdm` STRING COMMENT '',
        `jzdz` STRING COMMENT '',
        `fzrq` DATE COMMENT '',
        `dsc_biz_record_id` STRING COMMENT '',
        PRIMARY KEY (`dsc_biz_record_id`) NOT ENFORCED
) with (
        'connector' = 'stream-x'
);

然後提交任務的時候發現已經記錄了start-location 和 start-location的指標資訊了，但是並沒有上報到Prometheus！

在本地偵錯原始碼解決問題的大致過程

在類 com.dtstack.chunjun.source.format.BaseRichInputFormat中有一個成員變數

/** 自定義的prometheus reporter，用於提交startLocation和endLocation指標 */
protected transient CustomReporter customReporter;

該變數是用來提交增量資訊的物件，flink任務在開始的時候會執行一下方法

    @Override
    public void openInputFormat() throws IOException {
        Map<String, String> vars = getRuntimeContext().getMetricGroup().getAllVariables();
        if (vars != null) {
            jobName = vars.getOrDefault(Metrics.JOB_NAME, "defaultJobName");
            jobId = vars.get(Metrics.JOB_NAME);
            indexOfSubTask = Integer.parseInt(vars.get(Metrics.SUBTASK_INDEX));
        }

        LOG.info("是否使用自定義報告 {}", useCustomReporter());
        if (useCustomReporter()) {

            customReporter =
                    DataSyncFactoryUtil.discoverMetric(
                            config, getRuntimeContext(), makeTaskFailedWhenReportFailed());
            customReporter.open();
            LOG.info("customReporter 的hashcode is {}", customReporter.hashCode());
        }

        startTime = System.currentTimeMillis();
    }

通過排查useCustomReporter方法得知 jdbcConf.getInitReporter()是false，而在JdbcConfig類裡面這個物件預設是true

 /** 使用自定義的指標輸出器把增量指標打到普羅米修斯 */
    @Override
    protected boolean useCustomReporter() {
        return jdbcConf.isIncrement() && jdbcConf.getInitReporter();
    }

    /** 增量同步或者間隔輪詢時，是否初始化外部儲存 */
    protected Boolean initReporter = true;

經過查詢 initReporter 屬性的set方法呼叫，找到了下面的問題
在類 com.dtstack.chunjun.connector.jdbc.source.JdbcDynamicTableSource 中有個地方說暫時不支援SQL的方式

嘗試一下將false修改為true，然後在本地進行測試，測試的時候將pushgateway的host和port寫到程式碼裡面，執行任務發現pushgateway裡面已經有資料了

那麼可以開始打包了，由於改了原始碼，所以要先格式化程式碼 mvn spotless:apply 再打包 mvn clean package -DskipTests

後續問題

打包到虛擬機器器進行測試，我使用的是yarn-per-job模式，提交任務後發現報找不到Prometheus報告類的異常，通過異常資訊發現在前面提到的方法裡有classloader

public void openInputFormat() throws IOException {
        Map<String, String> vars = getRuntimeContext().getMetricGroup().getAllVariables();
        if (vars != null) {
            jobName = vars.getOrDefault(Metrics.JOB_NAME, "defaultJobName");
            jobId = vars.get(Metrics.JOB_NAME);
            indexOfSubTask = Integer.parseInt(vars.get(Metrics.SUBTASK_INDEX));
        }

        LOG.info("是否使用自定義報告 {}", useCustomReporter());
        if (useCustomReporter()) {

            customReporter =
                    DataSyncFactoryUtil.discoverMetric(
                            config, getRuntimeContext(), makeTaskFailedWhenReportFailed());
            customReporter.open();
            LOG.info("customReporter 的hashcode is {}", customReporter.hashCode());
        }

        startTime = System.currentTimeMillis();
    }

    public static CustomReporter discoverMetric(
            ChunJunCommonConf commonConf,
            RuntimeContext context,
            boolean makeTaskFailedWhenReportFailed) {
        try {
            String pluginName = commonConf.getMetricPluginName();
            // 這裡獲取到了類的全限定名 com.dtstack.chunjun.metrics.prometheus.PrometheusReport
            String pluginClassName = PluginUtil.getPluginClassName(pluginName, OperatorType.metric);
            MetricParam metricParam =
                    new MetricParam(
                            context, makeTaskFailedWhenReportFailed, commonConf.getMetricProps());

            ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
            Class<?> clazz = classLoader.loadClass(pluginClassName);
            Constructor<?> constructor = clazz.getConstructor(MetricParam.class);

            return (CustomReporter) constructor.newInstance(metricParam);
        } catch (Exception e) {
            throw new ChunJunRuntimeException(e);
        }
    }

在原生的時候這裡載入類的時候是沒問題的，但是線上上的時候出現了了找不到類的異常，猜測是相關的jar沒有載入到flink jvm程序裡面，所以將專案裡面的 chunjun-metrics-prometheus.jar 放到了flink的lib目錄下，再次啟動任務 問題得以解決！