@Test public void testStepSuffix() throws Exception { Set<Step> steps = Sets.newHashSet(); Config sourceStepConfig = ConfigFactory.empty(); BatchStep sourceStep = new BatchStep("source_step"); sourceStep.configure(sourceStepConfig); Dataset<Row> sourceDF = createTestDataframe(); sourceStep.setData(sourceDF); steps.add(sourceStep); Config loopStepConfig = ConfigFactory.empty() .withValue(Step.DEPENDENCIES_CONFIG, ConfigValueFactory.fromIterable(Lists.newArrayList("source_step"))) .withValue(LoopStep.MODE_PROPERTY, ConfigValueFactory.fromAnyRef(LoopStep.MODE_PARALLEL)) .withValue(LoopStep.SOURCE_PROPERTY, ConfigValueFactory.fromAnyRef(LoopStep.SOURCE_STEP)) .withValue(LoopStep.STEP_PROPERTY, ConfigValueFactory.fromAnyRef("source_step")) .withValue(LoopStep.SUFFIX_PROPERTY, ConfigValueFactory.fromAnyRef("descr")) ; RefactorStep loopStep = new LoopStep("loop_step"); loopStep.configure(loopStepConfig); steps.add(loopStep); Config step1Config = ConfigFactory.empty() .withValue(Step.DEPENDENCIES_CONFIG, ConfigValueFactory.fromIterable(Lists.newArrayList("loop_step"))) .withValue("sql", ConfigValueFactory.fromAnyRef("select ${non_existent_parm}, count(record_id) " + "from some_table " + "where record_month = '${descr}' " + "and record_date <> '${sdate}'")); Step step1 = new BatchStep("dependent_step"); step1.configure(step1Config); steps.add(step1); Set<Step> unrolled = loopStep.refactor(steps); assertEquals(15, unrolled.size()); assertEquals(StepUtils.getStepForName("source_step", unrolled).get().getDependencyNames(), Sets.newHashSet()); assertEquals(StepUtils.getStepForName("loop_step", unrolled).get().getDependencyNames(), Sets.newHashSet("source_step")); assertEquals(13, StepUtils.getImmediateDependentSteps(StepUtils.getStepForName("loop_step", unrolled).get(),unrolled).size()); assertEquals(StepUtils.getStepForName("dependent_step_January", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step")); assertTrue(StepUtils.getStepForName("dependent_step_January", unrolled).get().getConfig().getAnyRef("sql").toString() .matches(".+?\\$\\{non_existent_parm\\}.*?'January'.*?'2019-01-01'.*")); assertEquals(StepUtils.getStepForName("dependent_step_February", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step")); assertFalse(StepUtils.getStepForName("dependent_step_February", unrolled).get().getConfig().getAnyRef("sql").toString() .matches(".+?\\$\\{non_existent_parm\\}.*?'January'.*?'2019-02-01'.*")); assertEquals(StepUtils.getStepForName("dependent_step_null", unrolled).get().getDependencyNames(), Sets.newHashSet("loop_step")); assertTrue(StepUtils.getStepForName("dependent_step_null", unrolled).get().getConfig().getAnyRef("sql").toString() .matches(".+?\\$\\{non_existent_parm\\}.*?'null'.*?'9999-12-31'.*")); }
@Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } LoopStep(String name); }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } LoopStep(String name); @Override Set<Step> refactor(Set<Step> steps); @Override Step copy(); @Override Validations getValidations(); }
LoopStep extends RefactorStep implements ProvidesValidations { @Override public Set<Step> refactor(Set<Step> steps) { List<Row> iterationRows = getIterationRows(steps); String mode = getMode(); Set<Step> loopGraphSteps = StepUtils.getImmediateDependentSteps(this, steps); LOG.debug("Loop graph steps found: " + StepUtils.stepNamesAsString(loopGraphSteps)); Set<Step> loopGraphStepDependents = Sets.newHashSet(); for (Step loopGraphStep : loopGraphSteps) { for (Step loopGraphStepDependentCandidate : StepUtils.getImmediateDependentSteps(loopGraphStep, steps)) { if (!loopGraphSteps.contains(loopGraphStepDependentCandidate)) { loopGraphStepDependents.add(loopGraphStepDependentCandidate); } } } LOG.debug("Loop graph step dependents found: " + StepUtils.stepNamesAsString(loopGraphStepDependents)); steps.removeAll(loopGraphSteps); Set<Step> previousIterationSteps = null; for (Row iterationRow : iterationRows) { String suffix = getSuffix(iterationRow) ; LOG.debug("Constructing loop for iteration value: {}", suffix); Set<Step> iterationSteps = StepUtils.copySteps(loopGraphSteps); for (Step iterationStep : iterationSteps) { LOG.debug("Adjusting dependencies for iteration step: " + iterationStep.getName()); Set<String> dependencies = iterationStep.getDependencyNames(); Set<String> dependenciesToAdd = Sets.newHashSet(); Set<String> dependenciesToRemove = Sets.newHashSet(); for (String dependency : dependencies) { if (StepUtils.getStepForName(dependency, loopGraphSteps).isPresent()) { dependenciesToAdd.add(dependency + "_" + suffix); dependenciesToRemove.add(dependency); } } dependencies.addAll(dependenciesToAdd); LOG.debug("Added dependencies: " + dependenciesToAdd); dependencies.removeAll(dependenciesToRemove); LOG.debug("Removed dependencies: " + dependenciesToRemove); } for (Step iterationStep : iterationSteps) { String adjustedIterationStepName = iterationStep.getName() + "_" + suffix; LOG.debug("Renaming iteration step {} to {}", iterationStep.getName(), adjustedIterationStepName); iterationStep.setName(adjustedIterationStepName); for (Step loopGraphStepDependent : loopGraphStepDependents) { if (!(iterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to {}", iterationStep.getName(), loopGraphStepDependent.getName()); loopGraphStepDependent.getDependencyNames().add(adjustedIterationStepName); } } Config parameterSubstConfig = performSubstitutions(iterationStep.getConfig(), iterationRow); iterationStep.setConfig(parameterSubstConfig); LOG.debug("Parameter substitutions completed: {} ", parameterSubstConfig.toString()); } if (mode.equals(MODE_SERIAL) && previousIterationSteps != null) { LOG.debug("Adjusting iteration steps to be dependent on previous iteration's steps"); for (Step iterationStep : iterationSteps) { for (Step previousIterationStep : previousIterationSteps) { if (!(previousIterationStep instanceof LoopStep)) { LOG.debug("Adding dependency {} to iteration step {}", previousIterationStep.getName(), iterationStep.getName()); iterationStep.getDependencyNames().add(previousIterationStep.getName()); } } } } previousIterationSteps = iterationSteps; steps.addAll(iterationSteps); } for (Step loopGraphStepDependent : loopGraphStepDependents) { for (Step loopGraphStep : loopGraphSteps) { loopGraphStepDependent.getDependencyNames().remove(loopGraphStep.getName()); } } LOG.debug("Unrolled steps: " + StepUtils.stepNamesAsString(steps)); this.setState(StepState.FINISHED); return steps; } LoopStep(String name); @Override Set<Step> refactor(Set<Step> steps); @Override Step copy(); @Override Validations getValidations(); static final String MODE_PROPERTY; static final String MODE_SERIAL; static final String MODE_PARALLEL; static final String PARAMETER_PROPERTY; static final String SOURCE_PROPERTY; static final String SOURCE_RANGE; static final String SOURCE_LIST; static final String SOURCE_STEP; static final String RANGE_START_PROPERTY; static final String RANGE_END_PROPERTY; static final String LIST_PROPERTY; static final String STEP_PROPERTY; static final String SUFFIX_PROPERTY; static final String DEFAULT_SUFFIX_PROPERTY; }