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

feat(all): flink job增加tags属性

Browse Source 
flink job级别增加标签属性，用于区分调用测试包和非测试包
This commit is contained in:
v-zhangjc9
2024-07-30 16:42:02 +08:00
parent b0c5d04476
commit a3472340b5
23 changed files with 332 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/Synchronizer.java
View File

@@ -9,12 +9,16 @@ import com.lanyuanxiaoyao.service.common.entity.TableMeta;
import com.lanyuanxiaoyao.service.common.exception.CheckpointRootPathNotFoundException;
import com.lanyuanxiaoyao.service.common.exception.ZookeeperUrlNotFoundException;
import com.lanyuanxiaoyao.service.common.utils.NameHelper;
import com.lanyuanxiaoyao.service.common.utils.TableMetaHelper;
import com.lanyuanxiaoyao.service.common.utils.TagsHelper;
import com.lanyuanxiaoyao.service.sync.configuration.GlobalConfiguration;
import com.lanyuanxiaoyao.service.sync.functions.PulsarMessage2RecordFunction;
import com.lanyuanxiaoyao.service.sync.functions.PulsarMessageSourceReader;
import com.lanyuanxiaoyao.service.sync.functions.ValidateRecordFilter;
import com.lanyuanxiaoyao.service.sync.utils.*;
import com.lanyuanxiaoyao.service.sync.utils.ArgumentsUtils;
import com.lanyuanxiaoyao.service.sync.utils.JacksonUtils;
import com.lanyuanxiaoyao.service.sync.utils.StatusUtils;
import com.lanyuanxiaoyao.service.sync.utils.SyncUtils;
import com.lanyuanxiaoyao.service.sync.utils.ZkUtils;
import java.io.IOException;
import java.util.List;
import java.util.Map;
@@ -34,7 +38,10 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static com.lanyuanxiaoyao.service.common.Constants.*;
import static com.lanyuanxiaoyao.service.common.Constants.GB;
import static com.lanyuanxiaoyao.service.common.Constants.HOUR;
import static com.lanyuanxiaoyao.service.common.Constants.MINUTE;
import static com.lanyuanxiaoyao.service.common.Constants.TAGS_DISABLE_CHAINING;
/**
* 同步应用
@@ -92,7 +99,7 @@ public class Synchronizer {
environment.getCheckpointConfig().enableUnalignedCheckpoints();
environment.getCheckpointConfig().setTolerableCheckpointFailureNumber(5);
if (tableMetaList.stream().anyMatch(meta -> TableMetaHelper.existsTag(meta, TAGS_DISABLE_CHAINING))) {
if (tableMetaList.stream().anyMatch(meta -> TagsHelper.existsTag(meta, TAGS_DISABLE_CHAINING))) {
logger.warn("Disable operator chaining");
environment.disableOperatorChaining();
}
@@ -148,7 +155,7 @@ public class Synchronizer {
SingleOutputStreamOperator<String> source = environment
.addSource(new PulsarMessageSourceReader(configuration, flinkJob, tableMeta))
.setParallelism(tableMeta.getHudi().getSourceTasks());
if (TableMetaHelper.existsTag(tableMeta, Constants.TAGS_PULSAR_BACKUP)) {
if (TagsHelper.existsTag(tableMeta, Constants.TAGS_PULSAR_BACKUP)) {
Path path = new Path(StrUtil.format("hdfs://b2/apps/datalake/hive_test/source/{}/{}", String.join("_", flinkJob.getName().split("\\s")), tableMeta.getAlias()));
StreamingFileSink<String> fileSink = StreamingFileSink.<String>forRowFormat(path, new SimpleStringEncoder<>("UTF-8"))
.withRollingPolicy(DefaultRollingPolicy.builder()

2
utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/utils/ArgumentsUtils.java
View File

@@ -72,7 +72,7 @@ public class ArgumentsUtils {
return JacksonUtils.getMapper().readValue(argsTool.get(Constants.FLINK_JOB), FlinkJob.class);
}
public static String getInstants(String[] args) throws JsonProcessingException {
public static String getInstants(String[] args) {
ParameterTool argsTool = ParameterTool.fromArgs(args);
if (!argsTool.has(Constants.INSTANTS)) {
return "";

7
utils/sync/src/main/java/com/lanyuanxiaoyao/service/sync/utils/ConfigurationUtils.java
View File

@@ -7,6 +7,7 @@ import com.lanyuanxiaoyao.service.common.Constants;
import com.lanyuanxiaoyao.service.common.entity.FlinkJob;
import com.lanyuanxiaoyao.service.common.entity.TableMeta;
import com.lanyuanxiaoyao.service.common.utils.TableMetaHelper;
import com.lanyuanxiaoyao.service.common.utils.TagsHelper;
import com.lanyuanxiaoyao.service.sync.configuration.DefaultPartitionNameKeyGenerator;
import com.lanyuanxiaoyao.service.sync.configuration.GlobalConfiguration;
import java.util.Optional;
@@ -104,17 +105,17 @@ public class ConfigurationUtils {
configuration.setString(FlinkOptions.PATH, tableMeta.getHudi().getTargetHdfsPath());
configuration.setString(FlinkOptions.RECORD_KEY_FIELD, Constants.UNION_KEY_NAME);
configuration.setBoolean(FlinkOptions.PRE_COMBINE, false);
if (TableMetaHelper.existsTag(tableMeta, Constants.TAGS_PRE_COMBINE)) {
if (TagsHelper.existsTag(tableMeta, Constants.TAGS_PRE_COMBINE)) {
configuration.setBoolean(FlinkOptions.PRE_COMBINE, true);
}
configuration.setString(FlinkOptions.PRECOMBINE_FIELD, Constants.UPDATE_TIMESTAMP_KEY_NAME);
configuration.setString(FlinkOptions.SOURCE_AVRO_SCHEMA, schema.toString());
if (TableMetaHelper.existsTag(tableMeta, Constants.TAGS_NO_IGNORE_FAILED)) {
if (TagsHelper.existsTag(tableMeta, Constants.TAGS_NO_IGNORE_FAILED)) {
configuration.setBoolean(FlinkOptions.IGNORE_FAILED, false);
}
if (TableMetaHelper.existsTag(tableMeta, Constants.TAGS_DISABLE_HSYNC)) {
if (TagsHelper.existsTag(tableMeta, Constants.TAGS_DISABLE_HSYNC)) {
logger.info("Disable hsync");
configuration.setBoolean(HoodieWriteConfig.USE_HSYNC.key(), false);
}