基于chunjun纯钧的增量数据同步问题排查【博客园-实习小生】 ...

基于chunjun纯钧的增量数据同步

目前我司的大数据平台使用的是flink技术栈，底层的连接器插件使用的是国产的chunjun插件，在使用chunjun的过程中也遇到了很多问题，本次记录下在SQL模式的情况下怎么支持增量的数据同步

chunjun的官网文档对增量同步已经做出了一定的说明

纯钧官方
根据文档我编写了一个SQL脚本

1. create table `source` (
2.         `sfzh` STRING COMMENT '',
3.         `xm` STRING COMMENT '',
4.         `xb` STRING COMMENT '',
5.         `xbdm` STRING COMMENT '',
6.         `jzdz` STRING COMMENT '',
7.         `fzrq` DATE COMMENT '',
8.         `dsc_biz_record_id` STRING COMMENT ''
9. ) with (
10.         'connector' = 'mysql-x',
11.         'url' = 'jdbc:mysql://:/?useSSL=false&useInformationSchema=true&nullCatalogMeansCurrent=true',
12.         'table-name' = '',
13.         'username' = '',
14.         'password' = '',
15.         'scan.fetch-size' = '1024',
16.         'scan.increment.column' = 'fzrq',
17.         --'scan.increment.column-type' = 'date',
18.         'scan.start-location' = '1659974400000'
19. );
21. create table `sink` (
22.         `sfzh` STRING COMMENT '',
23.         `xm` STRING COMMENT '',
24.         `xb` STRING COMMENT '',
25.         `xbdm` STRING COMMENT '',
26.         `jzdz` STRING COMMENT '',
27.         `fzrq` DATE COMMENT '',
28.         `dsc_biz_record_id` STRING COMMENT '',
29.         PRIMARY KEY (`dsc_biz_record_id`) NOT ENFORCED
30. ) with (
31.         'connector' = 'stream-x'
32. );

复制代码

然后提交任务的时候发现已经记录了start-location 和 start-location的指标信息了，但是并没有上报到Prometheus！
在本地调试源码解决问题的大致过程

在类 com.dtstack.chunjun.source.format.BaseRichInputFormat中有一个成员变量

1. /** 自定义的prometheus reporter，用于提交startLocation和endLocation指标 */
2. protected transient CustomReporter customReporter;

复制代码

该变量是用来提交增量信息的对象，flink任务在开始的时候会执行一下方法

1.     @Override
2.     public void openInputFormat() throws IOException {
3.         Map<String, String> vars = getRuntimeContext().getMetricGroup().getAllVariables();
4.         if (vars != null) {
5.             jobName = vars.getOrDefault(Metrics.JOB_NAME, "defaultJobName");
6.             jobId = vars.get(Metrics.JOB_NAME);
7.             indexOfSubTask = Integer.parseInt(vars.get(Metrics.SUBTASK_INDEX));
8.         }
10.         LOG.info("是否使用自定义报告 {}", useCustomReporter());
11.         if (useCustomReporter()) {
13.             customReporter =
14.                     DataSyncFactoryUtil.discoverMetric(
15.                             config, getRuntimeContext(), makeTaskFailedWhenReportFailed());
16.             customReporter.open();
17.             LOG.info("customReporter 的hashcode is {}", customReporter.hashCode());
18.         }
20.         startTime = System.currentTimeMillis();
21.     }

复制代码

通过排查useCustomReporter方法得知 jdbcConf.getInitReporter()是false，而在JdbcConfig类里面这个对象默认是true

1. /** 使用自定义的指标输出器把增量指标打到普罗米修斯 */
2.     @Override
3.     protected boolean useCustomReporter() {
4.         return jdbcConf.isIncrement() && jdbcConf.getInitReporter();
5.     }
7.     /** 增量同步或者间隔轮询时，是否初始化外部存储 */
8.     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

1. public void openInputFormat() throws IOException {
2.         Map<String, String> vars = getRuntimeContext().getMetricGroup().getAllVariables();
3.         if (vars != null) {
4.             jobName = vars.getOrDefault(Metrics.JOB_NAME, "defaultJobName");
5.             jobId = vars.get(Metrics.JOB_NAME);
6.             indexOfSubTask = Integer.parseInt(vars.get(Metrics.SUBTASK_INDEX));
7.         }
9.         LOG.info("是否使用自定义报告 {}", useCustomReporter());
10.         if (useCustomReporter()) {
12.             customReporter =
13.                     DataSyncFactoryUtil.discoverMetric(
14.                             config, getRuntimeContext(), makeTaskFailedWhenReportFailed());
15.             customReporter.open();
16.             LOG.info("customReporter 的hashcode is {}", customReporter.hashCode());
17.         }
19.         startTime = System.currentTimeMillis();
20.     }
22.     public static CustomReporter discoverMetric(
23.             ChunJunCommonConf commonConf,
24.             RuntimeContext context,
25.             boolean makeTaskFailedWhenReportFailed) {
26.         try {
27.             String pluginName = commonConf.getMetricPluginName();
28.             // 这里获取到了类的全限定名 com.dtstack.chunjun.metrics.prometheus.PrometheusReport
29.             String pluginClassName = PluginUtil.getPluginClassName(pluginName, OperatorType.metric);
30.             MetricParam metricParam =
31.                     new MetricParam(
32.                             context, makeTaskFailedWhenReportFailed, commonConf.getMetricProps());
34.             ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
35.             Class<?> clazz = classLoader.loadClass(pluginClassName);
36.             Constructor<?> constructor = clazz.getConstructor(MetricParam.class);
38.             return (CustomReporter) constructor.newInstance(metricParam);
39.         } catch (Exception e) {
40.             throw new ChunJunRuntimeException(e);
41.         }
42.     }

复制代码

在本地的时候这里加载类的时候是没问题的，但是在线上的时候出现了了找不到类的异常，猜测是相关的jar没有加载到flink jvm进程里面，所以将项目里面的 chunjun-metrics-prometheus.jar 放到了flink的lib目录下，再次启动任务 问题得以解决！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！