stop the spark context with sparkContext.stop() and throw exception if the job fails
@@ -45,28 +45,23 @@ public class Validator {
if (log.isInfoEnabled()) {
log.info("Running Spark Validator with the following command line args (comma separated):{}", StringUtils.join(args, ","));
}
-
- try {
- final int result = new Validator().run(System.out, args);
- System.exit(result);
- } catch (final Exception e) {
- log.error("Failed to perform validation: ", e.toString(), e);
- System.out.println(e);
- System.exit(1);
- }
+ new Validator().run(System.out, args);
}
- private int run(@Nonnull final PrintStream out, @Nonnull final String... args) {
+ private void run(@Nonnull final PrintStream out, @Nonnull final String... args) {
// Check how many arguments were passed in
if (args.length < 4) {
- out.println("Proper Usage is: <targetDatabase> <entity> <partition> <path-to-policy-file>");
- out.println("You can optionally add: --hiveConf hive.setting=value --hiveConf hive.other.setting=value");
- out.println("You can optionally add: --storageLevel rdd_persistence_level_value");
- out.println("You can optionally add: --numPartitions number_of_rdd_partitions");
- out.println("You provided " + args.length + " args which are (comma separated): " + StringUtils.join(args, ","));
- return 1;
+ String msg = "Proper Usage is: <targetDatabase> <entity> <partition> <path-to-policy-file>\n" +
+ "You can optionally add: --hiveConf hive.setting=value --hiveConf hive.other.setting=value\n" +
+ "You can optionally add: --storageLevel rdd_persistence_level_value\n" +
+ "You can optionally add: --numPartitions number_of_rdd_partitions\n" +
+ "You provided " + args.length + " args which are (comma separated): " + StringUtils.join(args, ",");
+ out.println(msg);
+ throw new IllegalArgumentException(msg);
}
+ final SparkContext sparkContext = SparkContext.getOrCreate();
+ try {
final ValidatorConfiguration params = new ValidatorConfiguration(args);
// Initialize Spring context
@@ -74,7 +69,7 @@ public class Validator {
final DataValidator app = ctx.getBean(DataValidator.class);
// Prepare Hive context
- final HiveContext hiveContext = new HiveContext(SparkContext.getOrCreate());
+ final HiveContext hiveContext = new HiveContext(sparkContext);
for (final Param param : params.getHiveParams()) {
log.info("Adding Hive parameter {}={}", param.getName(), param.getValue());
@@ -95,7 +90,12 @@ public class Validator {
app.saveValidToTable(params.getTargetDatabase(), params.getFeedTableName(), params.getValidTableName(), results, hiveContext);
app.saveProfileToTable(params.getTargetDatabase(), params.getProfileTableName(), params.getPartition(), results, hiveContext);
results.unpersist();
-
- return 0;
+ log.info("Validator app finished");
+ } catch (Exception e) {
+ log.error("Failed to perform validation: {}", e.toString(), e);
+ throw e;
+ } finally {
+ sparkContext.stop();
+ }
}
}
