in hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java [108:311]
public static void main(String[] args) throws Exception {
ValidationUtils.checkArgument(args.length >= 4);
final String commandString = args[0];
LOG.info("Invoking SparkMain: " + commandString);
final SparkCommand cmd = SparkCommand.valueOf(commandString);
JavaSparkContext jsc = SparkUtil.initJavaSparkContext("hoodie-cli-" + commandString,
Option.of(args[1]), Option.of(args[2]));
int returnCode = 0;
try {
switch (cmd) {
case ROLLBACK:
assert (args.length == 6);
returnCode = rollback(jsc, args[3], args[4], Boolean.parseBoolean(args[5]));
break;
case DEDUPLICATE:
assert (args.length == 8);
returnCode = deduplicatePartitionPath(jsc, args[3], args[4], args[5], Boolean.parseBoolean(args[6]), args[7]);
break;
case ROLLBACK_TO_SAVEPOINT:
assert (args.length == 6);
returnCode = rollbackToSavepoint(jsc, args[3], args[4], Boolean.parseBoolean(args[5]));
break;
case IMPORT:
case UPSERT:
assert (args.length >= 13);
String propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[12])) {
propsFilePath = args[12];
}
List<String> configs = new ArrayList<>();
if (args.length > 13) {
configs.addAll(Arrays.asList(args).subList(13, args.length));
}
returnCode = dataLoad(jsc, commandString, args[3], args[4], args[5], args[6], args[7], args[8],
Integer.parseInt(args[9]), args[10], Integer.parseInt(args[11]), propsFilePath, configs);
break;
case COMPACT_RUN:
assert (args.length >= 10);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[9])) {
propsFilePath = args[9];
}
configs = new ArrayList<>();
if (args.length > 10) {
configs.addAll(Arrays.asList(args).subList(10, args.length));
}
returnCode = compact(jsc, args[3], args[4], args[5], Integer.parseInt(args[6]), args[7],
Integer.parseInt(args[8]), HoodieCompactor.EXECUTE, propsFilePath, configs);
break;
case COMPACT_SCHEDULE_AND_EXECUTE:
assert (args.length >= 9);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[8])) {
propsFilePath = args[8];
}
configs = new ArrayList<>();
if (args.length > 9) {
configs.addAll(Arrays.asList(args).subList(9, args.length));
}
returnCode = compact(jsc, args[3], args[4], null, Integer.parseInt(args[5]), args[6],
Integer.parseInt(args[7]), HoodieCompactor.SCHEDULE_AND_EXECUTE, propsFilePath, configs);
break;
case COMPACT_SCHEDULE:
assert (args.length >= 7);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[6])) {
propsFilePath = args[6];
}
configs = new ArrayList<>();
if (args.length > 7) {
configs.addAll(Arrays.asList(args).subList(7, args.length));
}
returnCode = compact(jsc, args[3], args[4], args[5], 1, "", 0, HoodieCompactor.SCHEDULE, propsFilePath, configs);
break;
case COMPACT_VALIDATE:
assert (args.length == 7);
doCompactValidate(jsc, args[3], args[4], args[5], Integer.parseInt(args[6]));
returnCode = 0;
break;
case COMPACT_REPAIR:
assert (args.length == 8);
doCompactRepair(jsc, args[3], args[4], args[5], Integer.parseInt(args[6]),
Boolean.parseBoolean(args[7]));
returnCode = 0;
break;
case COMPACT_UNSCHEDULE_FILE:
assert (args.length == 10);
doCompactUnscheduleFile(jsc, args[3], args[4], args[5], args[6], Integer.parseInt(args[7]),
Boolean.parseBoolean(args[8]), Boolean.parseBoolean(args[9]));
returnCode = 0;
break;
case COMPACT_UNSCHEDULE_PLAN:
assert (args.length == 9);
doCompactUnschedule(jsc, args[3], args[4], args[5], Integer.parseInt(args[6]),
Boolean.parseBoolean(args[7]), Boolean.parseBoolean(args[8]));
returnCode = 0;
break;
case CLUSTERING_RUN:
assert (args.length >= 9);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[8])) {
propsFilePath = args[8];
}
configs = new ArrayList<>();
if (args.length > 9) {
configs.addAll(Arrays.asList(args).subList(9, args.length));
}
returnCode = cluster(jsc, args[3], args[4], args[5], Integer.parseInt(args[6]), args[2],
Integer.parseInt(args[7]), EXECUTE, propsFilePath, configs);
break;
case CLUSTERING_SCHEDULE_AND_EXECUTE:
assert (args.length >= 8);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[7])) {
propsFilePath = args[7];
}
configs = new ArrayList<>();
if (args.length > 8) {
configs.addAll(Arrays.asList(args).subList(8, args.length));
}
returnCode = cluster(jsc, args[3], args[4], null, Integer.parseInt(args[5]), args[2],
Integer.parseInt(args[6]), SCHEDULE_AND_EXECUTE, propsFilePath, configs);
break;
case CLUSTERING_SCHEDULE:
assert (args.length >= 7);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[6])) {
propsFilePath = args[6];
}
configs = new ArrayList<>();
if (args.length > 7) {
configs.addAll(Arrays.asList(args).subList(7, args.length));
}
returnCode = cluster(jsc, args[3], args[4], args[5], 1, args[2],
0, SCHEDULE, propsFilePath, configs);
break;
case CLEAN:
assert (args.length >= 5);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[4])) {
propsFilePath = args[4];
}
configs = new ArrayList<>();
if (args.length > 5) {
configs.addAll(Arrays.asList(args).subList(5, args.length));
}
clean(jsc, args[3], propsFilePath, configs);
break;
case SAVEPOINT:
assert (args.length == 7);
returnCode = createSavepoint(jsc, args[3], args[4], args[5], args[6]);
break;
case DELETE_MARKER:
assert (args.length == 5);
returnCode = deleteMarker(jsc, args[3], args[4]);
break;
case DELETE_SAVEPOINT:
assert (args.length == 5);
returnCode = deleteSavepoint(jsc, args[3], args[4]);
break;
case BOOTSTRAP:
assert (args.length >= 18);
propsFilePath = null;
if (!StringUtils.isNullOrEmpty(args[17])) {
propsFilePath = args[17];
}
configs = new ArrayList<>();
if (args.length > 18) {
configs.addAll(Arrays.asList(args).subList(18, args.length));
}
returnCode = doBootstrap(jsc, args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10],
args[11], args[12], args[13], args[14], args[15], args[16], propsFilePath, configs);
break;
case UPGRADE:
case DOWNGRADE:
assert (args.length == 5);
returnCode = upgradeOrDowngradeTable(jsc, args[3], args[4]);
break;
case REPAIR_DEPRECATED_PARTITION:
assert (args.length == 4);
returnCode = repairDeprecatedPartition(jsc, args[3]);
break;
case RENAME_PARTITION:
assert (args.length == 6);
returnCode = renamePartition(jsc, args[3], args[4], args[5]);
break;
case ARCHIVE:
assert (args.length == 8);
returnCode = archive(jsc, Integer.parseInt(args[3]), Integer.parseInt(args[4]), Integer.parseInt(args[5]), Boolean.parseBoolean(args[6]), args[7]);
break;
default:
break;
}
} catch (Throwable throwable) {
LOG.error("Fail to execute commandString", throwable);
returnCode = -1;
} finally {
jsc.stop();
}
System.exit(returnCode);
}