lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-2032] Make keygen class and keygen type optional for FlinkStreamerConfig (#3104)

Browse Source 
* [HUDI-2032] Make keygen class and keygen type optional for FlinkStreamerConfig

* Address the review suggestion
This commit is contained in:
vinoyang
2021-06-17 21:22:13 +08:00
committed by GitHub
parent f97dd25d41
commit 67c3124352
1 changed files with 6 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-flink/src/main/java/org/apache/hudi/streamer/FlinkStreamerConfig.java
View File

@@ -21,6 +21,7 @@ package org.apache.hudi.streamer;
import org.apache.hudi.client.utils.OperationConverter; import org.apache.hudi.client.utils.OperationConverter;
import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload; import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
import org.apache.hudi.common.model.WriteOperationType; import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.keygen.constant.KeyGeneratorType; import org.apache.hudi.keygen.constant.KeyGeneratorType;
import org.apache.hudi.util.StreamerUtil; import org.apache.hudi.util.StreamerUtil;
@@ -155,6 +156,11 @@ public class FlinkStreamerConfig extends Configuration {
conf.setString(FlinkOptions.RECORD_KEY_FIELD, config.recordKeyField); conf.setString(FlinkOptions.RECORD_KEY_FIELD, config.recordKeyField);
conf.setString(FlinkOptions.PARTITION_PATH_FIELD, config.partitionPathField); conf.setString(FlinkOptions.PARTITION_PATH_FIELD, config.partitionPathField);
conf.setString(FlinkOptions.KEYGEN_CLASS, config.keygenClass); conf.setString(FlinkOptions.KEYGEN_CLASS, config.keygenClass);
if (!StringUtils.isNullOrEmpty(config.keygenClass)) {
conf.setString(FlinkOptions.KEYGEN_CLASS, config.keygenClass);
} else {
conf.setString(FlinkOptions.KEYGEN_TYPE, config.keygenType);
}
conf.setInteger(FlinkOptions.WRITE_TASKS, config.writeTaskNum); conf.setInteger(FlinkOptions.WRITE_TASKS, config.writeTaskNum);
return conf; return conf;