Email Ingestion and Deduplication

This repository provides a system for ingesting and deduplicating email files in real-time. It ensures that similar emails are grouped into canonical threads while maintaining hierarchical relationships between parent and child emails. The system is designed to work in a Kubernetes environment and performs real-time processing.

Assumptions

• All email files are in the same format
  • Most recent replies at the top
  • A second From: line marks the "parent" segment
• All email file names are unique (including future files yet to be ingested)
• Kubernetes cluster is set up and running
• Cloud-hosted PSQL database is up and running.
  • Alternatively, deploy a database on your k8s cluster
• User can connect to their database to run queries

High Level Approach

• Database setup: Two tables are created
  • canonical_thread: Maps every signature to a parent signature.
  • raw_mapping: Maps every document to its signature. Modified/duplicate documents will have the same signature.
• Signature generation: Each file’s full body (and, if present, its "parent" segment after a second From:) is normalized (lower-cased, punctuation/abbreviations replaced, non-alphanumerics stripped, whitespace removed) into a signature.
  • Look up the exact signature in canonical thread table, inserts it if new (handling race conditions), back-fills any missing parent link, and then records the raw mapping.
• Multi-pod approach: Pods discover .txt files in an input folder and atomically "claims" each file by moving it into a processing directory. Files are moved into a "processed" folder when done.
• Real-time construction: A continuous loop monitors the input folder for new .txt files. If no files are found, pods/workers temporarily sleep for 2 seconds before checking again.

Running locally

• Set up a PSQL database.
• Modify .env vars based on where documents are stored.
• Set PYTHONPATH -> export PYTHONPATH=$(pwd)
• Run python src/main.py.

K8s Deployment

Deploy your database (alternately, use a cloud-hosted PSQL db)

kubectl apply -f k8s/postgres.yaml

Wait for your database to enter a running state

Deploy email ingestor and deduplication pods

Replace the following fields in k8s/deployment.yaml with values relevant to your deployment:

• volumes.hostPath.path: The directory designated for dynamic email ingestion. This directory holds emails and supports real-time addition of new files. The container's /app/data folder will be mapped to this directory.
• DATABASE_URL: The URL of your database.

Then run the following commands:

docker build -t email-ingest:latest .
kubectl apply -f k8s/deployment.yaml

Evaluation

There are 70 documents in the eval directory. I verify that every document has been assigned to a canonical thread using the following query:

SELECT COUNT(*)
FROM raw_mapping
WHERE canon_signature IS NOT NULL;

The expected output is 70 (or the size of your test data)

Next, to get the relationships between canonical threads and original raw documents, run the following query:

SELECT
  array_agg(DISTINCT rm.doc_id ORDER BY rm.doc_id) AS raw_docs,
  array_agg(DISTINCT rmp.doc_id ORDER BY rmp.doc_id) AS parent_docs
FROM canonical_thread ct
LEFT JOIN raw_mapping rm
  ON rm.canon_signature = ct.signature
LEFT JOIN raw_mapping rmp
  ON rmp.canon_signature = ct.parent_signature
GROUP BY ct.signature;

The query results show a performance of 100% on the eval set. All similar emails are grouped together. We can also identify documents associated with parent signatures, thus maintaining hierarchy links between canonical threads.

For example, consider the last row of the query results:

• The raw documents 75_1_2m.txt, 75_1_2.txt, 75_1m_2m.txt, and 75_1m_2.txt are grouped together.
• The parent documents for this group are 75_1m.txt and 75_1.txt.
• The ultimate parent document for this group is 75.txt, which has no parent group (its parent signature is NULL).

Retrieve thread id (signature) given document id

SELECT canon_signature
FROM raw_mapping
WHERE doc_id = '24_1m.txt';

Retrieve set of document IDs give thread id (signature)

This query retrieves all documents that have the same signature as 24_1m.txt

SELECT doc_id
FROM raw_mapping
WHERE canon_signature = (
    SELECT canon_signature
    FROM raw_mapping
    WHERE doc_id = '24_1m.txt'
);

Retrieve parent id (signature) of thread

SELECT parent_signature
FROM canonical_thread ct
WHERE ct.signature = (
    SELECT canon_signature
    FROM raw_mapping
    WHERE doc_id = '89_1_2.txt'
);

Limitations

• Emails assumed to be in same format
• Processed files don't persist
  • Processed files are stored locally in each pod
• No hashing of signatures
• Signature isn't robust to modifications of alphanumeric characters
• No foreign key on parent signatures in case parents are processed after children
• File claiming is not robust, can use locking mechanism