Step 1: Hash-Based Deduplication for Exact Duplicates

This step builds a hash-based deduplication key from the serialized event. A byte-equivalent event produces the same key within the configured cache window.

The Goal

You will build a pipeline that generates a unique "fingerprint" (a hash) for each message, remembers the fingerprints it has seen recently, and drops any message whose fingerprint it has already seen.

The "Hash -> Cache -> Check -> Drop" Pattern

1. Hash: For each message, create a SHA-256 hash of its content. This creates a consistent, unique fingerprint.
2. Cache: Check whether this fingerprint has been seen before. threads: 1 serializes batches, and for_each completes each record's lookup, cache write, and duplicate drop before processing the next record in that batch.
3. Check: A mapping processor checks the result from the cache.
4. Drop: If the cache found a match, the message is deleted. Otherwise, the new fingerprint is cached and the message continues.

Implementation

1. Create the Deduplication Pipeline: Copy the following configuration into a file named deduplicator.yaml.

   Complete pipeline

   deduplicator.yaml

   name: hash-deduplicator
   description: A pipeline that removes exact duplicate messages.
   
   config:
     # 1. Define a cache resource to remember the hashes we've seen.
     cache_resources:
       - label: dedup_cache
         memory:
           default_ttl: 60s # Remember hashes for 60 seconds
   
     input:
       http_server:
         address: 0.0.0.0:8080
         path: /ingest
   
     pipeline:
       # Serialize cache checks so duplicate messages cannot race each other.
       threads: 1
       processors:
         # 2. HASH: Create a hash of the message content.
         - mapping: |
             root = this
             # format_json() ensures keys are sorted, creating a consistent hash
             root.dedup_hash = this.format_json().hash("sha256")
   
         - for_each:
             # 3. CACHE: Check whether we've seen this hash before.
             - branch:
                 request_map: root = this.dedup_hash
                 processors:
                   - cache:
                       resource: dedup_cache
                       operator: exists
                       key: '${!content()}'
                 result_map: root.is_duplicate = content().string().bool()
   
             # 4. CACHE NEW HASHES: Remember records that passed the lookup.
             - switch:
                 - check: '!this.is_duplicate'
                   processors:
                     - cache:
                         resource: dedup_cache
                         operator: set
                         key: '${!this.dedup_hash}'
                         value: seen
   
             # 5. DROP: Emit only the first occurrence.
             - mapping: |
                 root = if this.is_duplicate {
                   deleted()
                 } else {
                   this.merge({"status": "processed_new"})
                 }
   
     output:
       stdout:
         codec: lines

2. Deploy and Test:

   # --- Send the SAME message twice ---
   curl -X POST http://localhost:8080/ingest \
     -H "Content-Type: application/json" \
     -d '{"event_id": "abc-123", "message": "hello"}'
   
   curl -X POST http://localhost:8080/ingest \
     -H "Content-Type: application/json" \
     -d '{"event_id": "abc-123", "message": "hello"}'

3. Verify: The intended result retains the first message and drops the repeated hash. This page does not record an executed result.

Review serialization order, cache lifetime, and restart behavior before adapting this pattern.