☞ 商品介面的定義 價格、庫存量、發貨地點等。此外，它還可以提供商品的詳細信息，包括商品的圖片、詳細描述、規格參數、售後服務等。這些信息可以幫助用戶更好地瞭解商品，從而更好地選擇商品。 其次，電商平臺商品詳情介面的實現原理是基於RESTful API。RESTful API是一種基於HTTP協議的A ...
基於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目錄下，再次啟動任務 問題得以解決！
