I have one apache beam dataflow job written in java. That job reads some data flow DB and modify the result and publish that to pubsub
DB -> Update Object -> Publish
I want to capture the failure records while publishing in pubsub and write it to a separate file.
Is there any way to do this?
Thanks
There is not currently functionality to capture records that fail to be published to PubsubIO.Write. What sorts of failures would you like to fail permanently (vs being retried)?
You can catch all the errors from the code of the pipeline with TupleTags and try catch blocs on transformations.
Then you will have 2 sinks :
I created a library called Asgarde to simplify error handling with Beam :
https://github.com/tosun-si/asgarde
You can install this lib in your project with Maven (this version is linked to Beam 2.41.0) :
<dependency>
<groupId>fr.groupbees</groupId>
<artifactId>asgarde</artifactId>
<version>0.20.0</version>
</dependency>
or Gradle :
implementation group: 'fr.groupbees', name: 'asgarde', version: '0.20.0'
An example of usage in the pipeline :
// Input PCollection
final PCollection<String> wordPCollection....
final WithFailures.Result<PCollection<Integer>, Failure> resultComposer = CollectionComposer.of(wordPCollection)
.apply("Map", MapElements.into(TypeDescriptors.strings()).via((String word) -> word + "Test"))
.apply("FlatMap", FlatMapElements
.into(TypeDescriptors.strings())
.via((String line) -> Arrays.asList(Arrays.copyOfRange(line.split(" "), 1, 5))))
.apply("ParDo", MapElementFn.into(TypeDescriptors.integers()).via(word -> 1 / word.length()))
.getResult();
// Good output PCollection
PCollection<String> output = result.output();
// Bad failure PCollection
PCollection<Failure> failures = result.failures();
CollectionComposer proposed by Asgarde is created from a PCollection and behind the scene, each apply uses TupleTags and try catch bloc.
CollectionComposer returns a Result structure and from this, we can recover the output and failure PCollection
Your failure sink can be written to Bigquery GCS or other (dead letter queue).
You can also use native Beam if your prefer, you can check an example from my Github repository :