基于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目錄下，再次啟動任務 問題得以解決！

標籤：其他

上一篇：電商平臺商品詳情介面的應用場景

下一篇：云圖說｜云資料庫GaussDB如何做到卓越性能