CollectionComposer

CollectionComposer is the entry point of Asgarde. It wraps a PCollection, applies the steps of the flow with error handling, and keeps the failures of all the steps.

Create a composer

CollectionComposer.of(wordPCollection)

CollectionComposer.of(wordPCollection)

CollectionComposer.of(word_pcollection)

Chain the steps

Each step has a name, used as the Beam transform name and as the pipelineStep of its failures.

The apply method accepts:

the Beam MapElements and FlatMapElements: the composer adds exceptionsInto and exceptionsVia for you;
the Asgarde DoFn classes: MapElementFn, MapProcessContextFn, FlatMapElementFn, FlatMapProcessContextFn and FilterFn (see Transforms);
your own DoFn classes extending BaseElementFn (see Custom DoFn);
any PTransform returning a WithFailures.Result<PCollection<T>, Failure>.

CollectionComposer.of(input)
        .apply("Map", MapElements.into(TypeDescriptors.integers()).via((String word) -> 1 / word.length()))
        .apply("Filter", FilterFn.by(count -> count > 0))

MapElements and FlatMapElements with your own exceptionsInto/exceptionsVia are accepted too, as long as they produce Asgarde Failure objects:

CollectionComposer.of(input)
        .apply("Map", MapElements
                .into(TypeDescriptors.integers())
                .via((String word) -> 1 / word.length())
                .exceptionsInto(TypeDescriptor.of(Failure.class))
                .exceptionsVia(exElt -> Failure.from("Map", exElt)))

The Kotlin extensions give a concise syntax for each kind of step (see Kotlin extensions):

CollectionComposer.of(input)
    .map("Map") { word -> 1 / word.length }
    .filter("Filter") { count -> count > 0 }

The composer exposes the map, flat_map and filter operators (see Operators):

(CollectionComposer.of(input)
 .map('Map', lambda word: 1 / len(word))
 .filter('Filter', lambda count: count > 0))

Get the result

getResult() returns a Beam WithFailures.Result with the output of the last step and the failures of all the steps:

final WithFailures.Result<PCollection<Integer>, Failure> result = composer.getResult();

final PCollection<Integer> output = result.output();
final PCollection<Failure> failures = result.failures();

val result: Result<PCollection<Int>, Failure> = composer.result

val output: PCollection<Int> = result.output()
val failures: PCollection<Failure> = result.failures()

outputs = composer.outputs    # Output of the last step
failures = composer.failures  # Failures of all the steps