medium_ref_dead_letter_pattern.java

public class DeadLetterPatternExample {
    public static void main(String args[]) {
        DeadLetterPatternPipelineOptions pipelineOptions = PipelineOptionsFactory.fromArgs(args).withValidation()
            .as(DeadLetterPatternPipelineOptions.class);
        final Pipeline pipeline = org.apache.beam.sdk.Pipeline.create(pipelineOptions);

        PCollectionTuple parseMsg = pipeline
            .apply("PubSubReader",
                PubsubIO.readMessages().fromSubscription(pipelineOptions.getPubSubSubscriber()))
            .apply("ParseMsg",
            ParDo.of(new MsgParser())
                .withOutputTags(MsgParser.SuccessfulParse, TupleTagList.of(DeadLetterHandler.DeadLetterTag)));

        // Happy path - Write to BQ
        parseMsg
            .get(MsgParser.SuccessfulParse)
            .setCoder(TableRowJsonCoder.of())
            .apply("WriteToBQ",
                BigQueryIO.writeTableRows().to(pipelineOptions.getBqTable())
                    .withSchema(MsgParser.getTableSchema()));

        PCollection<TableRow> deadLetterErrorPCollection = parseMsg
            .get(DeadLetterHandler.DeadLetterTag)
            .setCoder(AvroCoder.of(DeadLetterHandler.DeadLetterError.class))
            .apply("BuildErrorRecord", ParDo.of(new DeadLetterHandler.BuildErrorRecord()));

        // Errors - Write raw data to BQ along with error details
        deadLetterErrorPCollection
            .apply("WriteErrorsToBQ",
                BigQueryIO.writeTableRows().to(pipelineOptions.getBqErrorsTable())
                    .withSchema(DeadLetterHandler.DeadLetterError.getTableSchema()));

        // Errors - Notify via slack on high error rates.
        deadLetterErrorPCollection
            .apply("ErrorsCounterWindow",
                Window.into(FixedWindows.of(Duration.standardSeconds(pipelineOptions.getErrorRateWindow()))))
            .apply("CountErrors", Combine.globally(new CountFn<TableRow>()).withoutDefaults())
            .setCoder(BigEndianLongCoder.of())
            .apply("NotifySlackOnHighErrorRate",
                ParDo.of(new DeadLetterHandler.NotifySlack(
                    pipelineOptions.getMaxErrorRate(),
                    pipelineOptions.getSlackKey(),
                    pipelineOptions.getSlackChannel(),
                    pipelineOptions.getJobName())));

        pipeline.run();
    }
}