lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-888] fix NullPointerException in HoodieCompactor (#1622)

Browse Source
This commit is contained in:
rolandjohann
2020-05-20 13:22:35 +02:00
committed by GitHub
parent 0dc2fa6172
commit 244d47494e
2 changed files with 19 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java
View File

@@ -270,7 +270,7 @@ public class SparkMain {
cfg.propsFilePath = propsFilePath; cfg.propsFilePath = propsFilePath;
cfg.configs = configs; cfg.configs = configs;
jsc.getConf().set("spark.executor.memory", sparkMemory); jsc.getConf().set("spark.executor.memory", sparkMemory);
return new HoodieCompactor(cfg).compact(jsc, retry); return new HoodieCompactor(jsc, cfg).compact(retry);
} }
private static int deduplicatePartitionPath(JavaSparkContext jsc, String duplicatedPartitionPath, private static int deduplicatePartitionPath(JavaSparkContext jsc, String duplicatedPartitionPath,

25
hudi-utilities/src/main/java/org/apache/hudi/utilities/HoodieCompactor.java
View File

@@ -43,11 +43,22 @@ public class HoodieCompactor {
private final Config cfg; private final Config cfg;
private transient FileSystem fs; private transient FileSystem fs;
private TypedProperties props; private TypedProperties props;
private final JavaSparkContext jsc;
public HoodieCompactor(Config cfg) { public HoodieCompactor(JavaSparkContext jsc, Config cfg) {
this.cfg = cfg; this.cfg = cfg;
this.props = cfg.propsFilePath == null ? UtilHelpers.buildProperties(cfg.configs) this.jsc = jsc;
: UtilHelpers.readConfig(fs, new Path(cfg.propsFilePath), cfg.configs).getConfig(); this.props = cfg.propsFilePath == null
? UtilHelpers.buildProperties(cfg.configs)
: readConfigFromFileSystem(jsc, cfg);
}
private TypedProperties readConfigFromFileSystem(JavaSparkContext jsc, Config cfg) {
final FileSystem fs = FSUtils.getFs(cfg.basePath, jsc.hadoopConfiguration());
return UtilHelpers
.readConfig(fs, new Path(cfg.propsFilePath), cfg.configs)
.getConfig();
} }
public static class Config implements Serializable { public static class Config implements Serializable {
@@ -90,12 +101,12 @@ public class HoodieCompactor {
cmd.usage(); cmd.usage();
System.exit(1); System.exit(1);
} }
HoodieCompactor compactor = new HoodieCompactor(cfg); final JavaSparkContext jsc = UtilHelpers.buildSparkContext("compactor-" + cfg.tableName, cfg.sparkMaster, cfg.sparkMemory);
compactor.compact(UtilHelpers.buildSparkContext("compactor-" + cfg.tableName, cfg.sparkMaster, cfg.sparkMemory), HoodieCompactor compactor = new HoodieCompactor(jsc, cfg);
cfg.retry); compactor.compact(cfg.retry);
} }
public int compact(JavaSparkContext jsc, int retry) { public int compact(int retry) {
this.fs = FSUtils.getFs(cfg.basePath, jsc.hadoopConfiguration()); this.fs = FSUtils.getFs(cfg.basePath, jsc.hadoopConfiguration());
int ret = -1; int ret = -1;
try { try {