Dead Letter Queue (DLQ)

1. Introduction

2. Why This Concept Exists

Dropping failed records silently is often unacceptable for auditing purposes. Routing them to a DLQ alongside their original payload and stack trace allows developers to analyze errors without halting the primary high-throughput processing pipeline.

3. Code Example

Using multi-output tagging to implement a DLQ pattern:

pythonCopy
import apache_beam as beam

class ParseTransactionDoFn(beam.DoFn):
    OUTPUT_SUCCESS = "success"
    OUTPUT_FAILED = "failed"

    def process(self, element):
        try:
            parts = element.split(",")
            tx_id = parts[0].strip()
            amount = float(parts[1].strip())
            yield beam.pvalue.TaggedOutput(self.OUTPUT_SUCCESS, (tx_id, amount))
        except Exception as e:
            yield beam.pvalue.TaggedOutput(self.OUTPUT_FAILED, {"raw_payload": element, "error": str(e)})

with beam.Pipeline() as p:
    results = (p 
               | beam.Create(["TX_100,50.5", "TX_200,invalid_amount", "TX_300,10.2"])
               | "Parse" >> beam.ParDo(ParseTransactionDoFn()).with_outputs(
                   ParseTransactionDoFn.OUTPUT_SUCCESS, 
                   ParseTransactionDoFn.OUTPUT_FAILED
               ))

    # Route successful output
    results.success | "PrintSuccess" >> beam.Map(lambda x: print(f"Success: {x}"))
    # Route failed output to DLQ
    results.failed | "PrintDLQ" >> beam.Map(lambda x: print(f"DLQ Alert: {x}"))

4. Key Takeaways

• Use beam.pvalue.TaggedOutput inside a DoFn to split processing streams.
• Write DLQ payloads to durable storage (like GCS) with partition directories reflecting execution dates.