lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(sync): 修复指标名称重复

Browse Source
This commit is contained in:
v-zhangjc9
2024-07-30 19:06:01 +08:00
parent d8b06e4232
commit 63b81155b8
2 changed files with 10 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
service-common/src/main/java/com/lanyuanxiaoyao/service/common/Constants.java
View File

@@ -101,6 +101,11 @@ public interface Constants {
String METRICS_SYNC_SOURCE_MESSAGE_RECEIVE = METRICS_SYNC_PREFIX + "_source_message_receive";
String METRICS_SYNC_SOURCE_MESSAGE_SIZE_RECEIVE_BYTES = METRICS_SYNC_PREFIX + "_source_message_receive_bytes";
String METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE = METRICS_SYNC_PREFIX + "_source_operation_type_receive";
String METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_INSERT = METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE + "_insert";
String METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_UPDATE = METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE + "_update";
String METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_DELETE = METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE + "_delete";
String METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_DDL = METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE + "_ddl";
String METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_UNKNOWN = METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE + "_unknown";
String METRICS_SYNC_SOURCE_CHANGE_FILTER = METRICS_SYNC_PREFIX + "_source_change_filter";
String METRICS_SYNC_SOURCE_CHANGE_PARTITION = METRICS_SYNC_PREFIX + "_source_change_partition";
String METRICS_SYNC_SOURCE_BACK_LOGS = METRICS_SYNC_PREFIX + "_source_back_logs";

13
utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/functions/OperationTypeFilter.java
View File

@@ -38,15 +38,12 @@ public class OperationTypeFilter extends RichFilterFunction<Record> {
public void open(Configuration parameters) throws Exception {
super.open(parameters);
Function<String, Map<String, String>> fillTags = operator -> MapUtil.<String, String>builder()
.put(Constants.METRICS_LABEL_TYPE, operator)
.build();
// 初始化指标
insertCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE, fillTags.apply(Constants.INSERT));
updateCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE, fillTags.apply(Constants.UPDATE));
deleteCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE, fillTags.apply(Constants.DELETE));
ddlCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE, fillTags.apply(Constants.DDL));
unknownCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE, fillTags.apply(Constants.UNKNOWN));
insertCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_INSERT);
updateCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_UPDATE);
deleteCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_DELETE);
ddlCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_DDL);
unknownCounter = MetricsUtils.counter(getRuntimeContext(), flinkJob, tableMeta, Constants.METRICS_SYNC_SOURCE_OPERATION_TYPE_RECEIVE_UNKNOWN);
}
@Override