Wsrtjtyk

Using Apache Beam I am doing computations - and if they succeed I'd like to write the output to one sink, and if there is a failure I'd like to write that to another sink.

Is there any way to handle metadata or content based routing in Apache Beam?

I've used Apache Camel extensively, and so in my mind based on the outcome of a previous transform, I should route a message to a different sink using a router (perhaps determined by a metadata flag I set on the message header). Is there an analogous capability with Apache Beam, or would I instead just have a sequential transform that inspects the PCollection and handles writing to sinks within the transform?

Ideally I'd like this logic (written verbosely for attempted clarity)

result = my_pcollections | 'compute_stuff' >> beam.Map(lambda (pcollection): my_compute_func(pcollection))

result | ([success_failure_router]

   | 'sucess_sink' >> beam.io.WriteToText('/path/to/file')

   | 'failure_sink' >> beam.io.WriteStringsToPubSub('mytopic'))

However.. I suspect the 'Beam' way of handling this is

result = my_pcollections | 'compute_stuff' >> beam.Map(lambda (pcollection): my_compute_func(pcollection))

result | 'write_results_appropriately' >> write_results_appropriately(result))

...

def write_results_appropriately(result):

   if result == ..:

      # success, write to file

   else:

      # failure, write to topic

Thanks,
Kevin

High-level:

I am not sure of specifics of the Python API in this case, but from high level it looks like this:

• par-dos support multiple outputs;


• outputs are identified by the tag you give them (e.g. "correct-elements", "invalid-elements");


• in your main par-do you write to multiple outputs choosing the output using your criteria;


• each output is represented by a separate PCollection;


• then you get the separate PCollections representing the tagged outputs from your par-do;


• then apply different sinks to each of the tagged PCollections;

In detail see the section
https://beam.apache.org/documentation/programming-guide/#additional-outputs