Your pipelines copy personal data faster than any human can track it. Learn how to scrub PII in flight — Kafka consumers, Spark UDFs, and Airflow tasks that detect and mask sensitive data before it lands anywhere new.
Build the ScrubberEvery copy of personal data your organization holds was put there by a pipeline. The CRM row that ended up in the warehouse, the clickstream event that landed in the lake, the support transcript that reached the search index and the ML feature store — all of them traveled through Kafka topics, Spark jobs, or orchestrated batch loads. That makes the pipeline layer uniquely powerful for privacy engineering: it is the one place where data is already in motion, already being transformed, and already passing through code you control. Detection applied here protects every downstream consumer at once, instead of being re-implemented in each warehouse, dashboard, and model that eventually touches the data.
Contrast this with the alternatives. Scanning data at rest — the approach described in our database discovery guide — finds PII after it has already spread, and each new destination must be scanned separately. Application-level controls catch data at the front door but miss everything generated internally: agent notes, enrichment payloads, third-party feeds. A pipeline scrubber sits between source and destination, so a single integration point turns raw topics into clean topics and raw staging tables into masked production tables. Downstream teams then inherit privacy by default rather than by policy.
The mechanics are straightforward because detection is just an HTTP call. Each record's text fields are posted to the PII Detection API, which returns typed entities with offsets and confidence scores plus an optional masked rendition of the input. The pipeline swaps the raw field for the anonymized_text value, attaches the entity metadata for auditing, and moves on. One request handles any of 150+ entity types in 60+ languages, so the same scrubber works for a German invoice feed and an English chat stream:
curl -X POST https://piidetectionapi.com/api/moderate.php \
-H "Content-Type: application/json" \
-d '{
"api_key": "YOUR_API_KEY",
"api_type": "pii_detection",
"text": "Order 8841: refund to Priya Sharma, card 4111 1111 1111 1111, call +44 7700 900123",
"entities": ["PERSON_NAME", "CREDIT_CARD_NUMBER", "PHONE_NUMBER"],
"mask_mode": "replace",
"threshold": 0.6
}'The response carries both the evidence (detected_entities with start/end offsets and confidence) and the remediation (anonymized_text reading "Order 8841: refund to [NAME], card [CREDIT_CARD], call [PHONE]"). Everything else in this guide is about wiring that call into Kafka, Spark, and Airflow at production throughput without breaking delivery guarantees. You can paste the same payload into the interactive demo to see the full response shape before writing any pipeline code.
The first architectural decision is not which framework but when detection happens relative to data movement. Streaming scrubbing processes each event as it flows through a topic or stream, so sensitive fields are masked seconds after creation and nothing downstream ever sees raw PII. This is the strongest privacy posture — the clean topic becomes the only topic most consumers are allowed to read — and it is the natural fit when downstream latency requirements are tight: fraud features, live dashboards, RAG ingestion (see our RAG pipeline guide), or webhook fan-out.
The costs of streaming are operational. The detection call is now on the delivery path, so its latency, error handling, and rate limits shape your consumer lag; you need dead-letter routing for records that repeatedly fail; and per-event calls forgo some batching efficiency unless you buffer micro-batches inside the consumer. Streaming scrubbers are long-running services that need deployment, monitoring, and scaling like any other stateful consumer group.
Batch scrubbing runs detection as a discrete step in a scheduled job: land raw data in a quarantined staging zone, scan and mask it, then publish to the trusted zone. Batching amortizes beautifully — thousands of records can be packed into large requests, retries are simple re-runs, and failures block publication rather than corrupt it. The price is a window of exposure: between landing and scrubbing, raw PII exists in staging, and everything reading staging must be locked down. Batch fits nightly warehouse loads, historical backfills, and any ELT pattern where the transform step already exists.
Micro-batching — Spark Structured Streaming triggers, Kafka consumers that poll hundreds of records at a time, or five-minute incremental jobs — is the pragmatic middle. You keep near-real-time freshness while sending fewer, fuller API requests. In practice most production deployments converge here: even "streaming" consumers should accumulate a poll's worth of records into one detection request rather than calling per event.
The three execution models differ less in what they detect than in what they promise operationally. Use this table to match the model to each data flow — a single organization will typically run all three against the same detection endpoint.
| Dimension | Streaming (per event) | Micro-Batch | Batch (scheduled) |
|---|---|---|---|
| Detection latency | Sub-second to seconds; PII masked before any consumer reads it | Seconds to minutes, bounded by poll/trigger interval | Minutes to hours; exposure window in staging until job runs |
| Throughput profile | Limited by per-request overhead; scale via consumer-group parallelism | High — each request carries 50–200 records up to the 50,000-char limit | Highest — full parallel fan-out across workers, easy rate-limit planning |
| Cost per record | Highest if called per event; drops sharply with in-consumer buffering | Low — batching amortizes request overhead across the poll | Lowest — maximal packing plus off-peak scheduling |
| Failure semantics | Needs DLQ + offset management; a stuck record can stall a partition | Retry the micro-batch; DLQ individual poison records | Rerun the job; failures block publication (naturally fail-closed) |
| Operational burden | Always-on service: scaling, lag monitoring, deploys | Same service, but simpler tuning and fewer requests | Just another task in the orchestrator with normal alerting |
| Best fit | Payment events, chat/LLM inputs, webhook fan-out, CDC streams | Clickstream, log enrichment, near-real-time warehouse sinks | Nightly ELT loads, historical backfills, third-party file drops |
One number worth internalizing when reading the throughput row: a typical detection request over a packed 50,000-character payload returns in a few hundred milliseconds. At 300 characters per record that is roughly 160 records per request — so a single worker sustaining four concurrent requests clears over 2,000 records per second, and batch jobs simply multiply workers. The bottleneck in well-built scrubbers is almost never the API; it is unbatched, per-event calling, which is the pattern the code below avoids.
The canonical streaming pattern is a scrubber consumer group sitting between a raw topic and a clean topic. Producers keep writing unmodified events to events.raw, which is locked down to the scrubber alone; every other consumer reads events.clean. The scrubber polls a batch of records, packs their text fields into one detection request with newline separators, splits the returned anonymized_text back into per-record masked values, and produces downstream. Records that fail after retries go to a dead-letter topic instead of blocking the partition:
import json, requests
from kafka import KafkaConsumer, KafkaProducer
API_URL = "https://piidetectionapi.com/api/moderate.php"
API_KEY = "YOUR_API_KEY"
SEP = "\n<__REC__>\n" # separator no real message contains
consumer = KafkaConsumer(
"events.raw",
group_id="pii-scrubber",
bootstrap_servers="kafka:9092",
enable_auto_commit=False,
value_deserializer=lambda b: json.loads(b.decode()),
)
producer = KafkaProducer(
bootstrap_servers="kafka:9092",
value_serializer=lambda v: json.dumps(v).encode(),
)
def scrub_batch(texts):
# One API call for the whole poll: pack, detect, unpack
resp = requests.post(API_URL, json={
"api_key": API_KEY,
"api_type": "pii_detection",
"text": SEP.join(texts),
"mask_mode": "replace",
"threshold": 0.6,
}, timeout=30)
resp.raise_for_status()
data = resp.json()
return data["anonymized_text"].split(SEP), data["entities_detected"]
while True:
polled = consumer.poll(timeout_ms=1000, max_records=100)
for tp, records in polled.items():
texts = [r.value.get("message", "")[:2000] for r in records]
try:
masked, n_found = scrub_batch(texts)
for rec, clean_text in zip(records, masked):
event = dict(rec.value, message=clean_text, pii_scrubbed=True)
producer.send("events.clean", event)
except Exception as exc:
# Fail-closed: raw events go to the DLQ, never to the clean topic
for rec in records:
producer.send("events.deadletter",
{"error": str(exc), "offset": rec.offset,
"partition": tp.partition, "payload": rec.value})
producer.flush()
consumer.commit() # commit only after clean/DLQ produce succeedsThree details carry the production weight here. Offsets are committed manually after the produce succeeds, so a crash mid-batch replays records rather than losing them — at-least-once delivery with idempotent downstream handling. The record separator lets one request scrub the whole poll while keeping a per-record mapping. And the exception path is fail-closed: on detection failure, raw text is quarantined in the DLQ rather than leaked to events.clean. The same topology in Node.js using kafkajs-style APIs:
import { Kafka } from "kafkajs";
const kafka = new Kafka({ brokers: ["kafka:9092"] });
const consumer = kafka.consumer({ groupId: "pii-scrubber-js" });
const producer = kafka.producer();
async function scrub(text) {
const resp = await fetch("https://piidetectionapi.com/api/moderate.php", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
api_key: process.env.PII_API_KEY,
api_type: "pii_detection",
text,
mask_mode: "replace",
threshold: 0.6,
}),
});
if (!resp.ok) throw new Error(`API ${resp.status}`);
return resp.json();
}
await consumer.connect();
await producer.connect();
await consumer.subscribe({ topic: "events.raw" });
await consumer.run({
eachBatch: async ({ batch, resolveOffset, heartbeat }) => {
for (const msg of batch.messages) {
const event = JSON.parse(msg.value.toString());
try {
const { anonymized_text, detected_entities } = await scrub(event.message);
await producer.send({
topic: "events.clean",
messages: [{ value: JSON.stringify({
...event,
message: anonymized_text,
pii_types: [...new Set(detected_entities.map(e => e.type))],
}) }],
});
} catch (err) {
await producer.send({
topic: "events.deadletter",
messages: [{ value: JSON.stringify({ error: String(err), event }) }],
});
}
resolveOffset(msg.offset);
await heartbeat();
}
},
});In Spark, detection becomes a column transformation. The efficient shape is a pandas_udf, because it hands your function whole Arrow batches instead of single rows — which maps perfectly onto packing many values into one API request. The UDF below masks a free-text column across however many executors the job runs, with each executor batching its own partitions' values:
import pandas as pd
import requests
from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType
API_URL = "https://piidetectionapi.com/api/moderate.php"
API_KEY = "YOUR_API_KEY"
SEP = "\n<__REC__>\n"
CHUNK_CHARS = 45_000 # stay under the 50,000-char request limit
def _scrub_values(values):
masked, chunk, size = [], [], 0
def flush():
nonlocal chunk, size
if not chunk:
return
r = requests.post(API_URL, json={
"api_key": API_KEY,
"api_type": "pii_detection",
"text": SEP.join(chunk),
"mask_mode": "replace",
"threshold": 0.6,
}, timeout=60)
r.raise_for_status()
masked.extend(r.json()["anonymized_text"].split(SEP))
chunk, size = [], 0
for v in values:
v = (v or "")[:5000]
if size + len(v) > CHUNK_CHARS:
flush()
chunk.append(v); size += len(v) + len(SEP)
flush()
return masked
@pandas_udf(StringType())
def mask_pii(col: pd.Series) -> pd.Series:
return pd.Series(_scrub_values(col.tolist()), index=col.index)
spark = SparkSession.builder.appName("pii-scrub").getOrCreate()
df = spark.read.parquet("s3://lake/raw/support_tickets/dt=2026-08-25/")
clean = df.withColumn("agent_notes", mask_pii("agent_notes")) \
.withColumn("customer_msg", mask_pii("customer_msg"))
clean.write.mode("overwrite").parquet("s3://lake/clean/support_tickets/dt=2026-08-25/")Two Spark-specific cautions. First, control total parallelism: every executor calling the API concurrently multiplies your request rate, so either cap executor count for the scrub stage, or coalesce the DataFrame to a partition count that matches the concurrency your plan tier allows. Second, make the job idempotent — write to a dated clean partition and overwrite on retry, so a mid-job executor failure never leaves half-masked data presented as fully scrubbed. If you need the entity metadata rather than just masked text, return a struct from the UDF containing both anonymized_text and a JSON-encoded entity list, and store it alongside the clean column for audit queries.
In an orchestrated ELT world, scrubbing is simply a task in the DAG: land raw data, scrub staging, verify, publish. Placing detection between load and publish gives you a natural gate — if the scrub task fails, the publish task never runs, and yesterday's clean data keeps serving. The task below scans a daily staging table, masks flagged rows in place, and pushes scan metrics to XCom so a downstream sensor can enforce a "zero unmasked criticals" rule:
from datetime import datetime
import requests
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
API_URL = "https://piidetectionapi.com/api/moderate.php"
CRITICAL = {"CREDIT_CARD_NUMBER", "SSN", "IBAN_CODE", "PASSWORD"}
def scrub_staging(ds, **ctx):
hook = PostgresHook(postgres_conn_id="warehouse")
rows = hook.get_records(
"SELECT id, notes FROM staging.orders WHERE load_date = %s", parameters=(ds,))
stats = {"rows": len(rows), "entities": 0, "critical": 0}
for i in range(0, len(rows), 100):
batch = rows[i:i+100]
sep = "\n<__REC__>\n"
resp = requests.post(API_URL, json={
"api_key": "YOUR_API_KEY",
"api_type": "pii_detection",
"text": sep.join((r[1] or "")[:2000] for r in batch),
"mask_mode": "replace",
"threshold": 0.55,
}, timeout=60)
resp.raise_for_status()
data = resp.json()
stats["entities"] += data["entities_detected"]
stats["critical"] += sum(
1 for e in data["detected_entities"] if e["type"] in CRITICAL)
for (row_id, _), clean in zip(batch, data["anonymized_text"].split(sep)):
hook.run("UPDATE staging.orders SET notes = %s, pii_scrubbed = true "
"WHERE id = %s", parameters=(clean, row_id))
ctx["ti"].xcom_push(key="scrub_stats", value=stats)
with DAG(
dag_id="orders_elt",
start_date=datetime(2026, 1, 1),
schedule="@daily",
catchup=False,
) as dag:
scrub = PythonOperator(
task_id="scrub_pii",
python_callable=scrub_staging,
retries=3,
retry_exponential_backoff=True,
)
# load_raw >> scrub >> verify_zero_criticals >> publish_to_martsThe orchestrator's native machinery does the heavy lifting: retries with exponential backoff absorbs transient API or network errors, task-level SLAs alert when a scrub runs long, and the XCom stats feed both dashboards and the gating task. For very large days, swap the in-task loop for a SparkSubmitOperator that runs the UDF job from the previous section — the DAG shape stays identical. Teams already scanning logs with a similar pattern should compare notes with our log scanning guide, which applies the same gate to log shippers.
max_active_runs and pool limits so backfills respect the same API concurrency budget as the daily run, or you will discover your rate limit at 2 a.m.Detection at pipeline scale is an exercise in request shaping. Four levers determine whether your scrubber clears millions of records an hour or becomes the slowest stage in the DAG.
The single biggest win: fill each request toward the 50,000-character ceiling using a sentinel separator, then split the masked output. Moving from one call per record to ~150 records per call cuts request count by two orders of magnitude, which shrinks both cost and total wall-clock time. Track your average characters-per-request as a first-class pipeline metric — drift downward means someone reintroduced per-event calls.
Run a small pool of in-flight requests per worker (4–8 is typical) behind a semaphore, and let the pool's saturation propagate backward: when detection slows, the Kafka consumer polls less and lag grows visibly instead of memory growing invisibly. Never buffer unbounded records awaiting scrubbing — that queue is itself an unmanaged store of raw PII. React to HTTP 429s with exponential backoff and jitter rather than hammering the retry.
For streaming paths, set an explicit p99 budget for added latency — say, 500 ms — and measure the scrubber against it. The API reports its own processing_time_ms per response, letting you separate model time from network and queueing time. Scope requests with the entities parameter when a flow only needs a handful of types; a narrower detection target returns faster and yields fewer false positives to handle downstream.
Pipelines re-deliver: at-least-once semantics, replays, and backfills mean the same payload arrives repeatedly. Cache scrub results keyed by a content hash so replayed records skip the API entirely, and skip fields that are structurally incapable of holding PII (numeric IDs, enums, booleans). Many teams cut 20–40% of request volume with a hash cache alone — deduplication is the cheapest capacity you will ever buy.
A scrubber's failure behavior is a privacy decision, not just a reliability one. Decide deliberately what happens to a record the pipeline could not scan — because that record might be the one carrying a card number.
Fail-closed means an unscannable record never reaches the clean destination — it waits in quarantine until detection succeeds. Fail-open means the pipeline forwards the raw record and logs the miss. Fail-closed is the only defensible default for regulated data: an outage then costs you freshness, not compliance. Reserve fail-open for flows where availability genuinely outranks exposure, and even then tag forwarded records (pii_scrubbed: false) so downstream can filter them.
Dead-letter messages contain exactly the data that failed to get masked — the DLQ is by definition your most PII-dense topic. Encrypt it, restrict read access to the reprocessing service, set a short retention, and never wire the DLQ into general-purpose log viewers. Include enough context to reprocess (source topic, partition, offset, error, attempt count) and reprocess through the same scrubbing path, not a shortcut that bypasses detection.
Distinguish transient failures (timeouts, 429s, network) from permanent ones (malformed payloads, oversize fields, encoding garbage). Transients deserve in-place retries with capped exponential backoff; permanents should go to the DLQ on the first attempt, because retrying a poison message stalls the partition behind it. A message that exhausts, say, five attempts across ten minutes graduates to poison regardless of error class. Log the API's status code with every failure — a spike of 4xx errors means a producer changed its payload shape, not that the network is flaky.
Patterns that keep a scrubber trustworthy after the first incident, the first audit, and the first 10x traffic spike.
A scrubber only helps if consumers cannot route around it. Use Kafka ACLs, warehouse grants, and bucket policies so that raw topics and staging zones are readable exclusively by the scrubbing service. Every downstream credential should physically lack access to unscrubbed data — then a new team's misconfigured job fails loudly at connect time instead of silently ingesting raw PII.
Publish per-flow counters: records scanned, entities found by type, mean confidence, DLQ depth, API latency percentiles. Entity-rate anomalies are leak detectors — a sudden jump in CREDIT_CARD_NUMBER detections in a telemetry topic means an upstream service started logging something it shouldn't. Route those alerts to the owning team with the same severity as an error-rate spike; the masked payloads themselves are safe to sample into dashboards as evidence.
Thresholds, entity scopes, mask_mode, and custom_instruction strings are policy — keep them in reviewed configuration, stamp each scrubbed record with the config version that processed it, and re-run affected windows when policy tightens. When an auditor asks "was this topic masked for SSNs in March?", the answer should be a config-version lookup, not archaeology. Validating that a threshold change helps rather than hurts is exactly what our accuracy measurement guide covers.
processing_time_ms field shows the model's share, with the remainder being network. Because a well-built consumer scrubs an entire poll (100+ records) in one call, the amortized per-record cost is single-digit milliseconds. End-to-end, most streaming scrubbers add well under a second of pipeline latency at p99, which fits comfortably inside typical freshness SLAs for dashboards, feature stores, and LLM ingestion.mask_mode: "replace" preserves the field as a string. When schemas evolve, new string fields should default to "scan until proven clean," not the reverse.One endpoint turns raw topics into clean topics — Kafka, Spark, Airflow, or anything that can make an HTTP call. Test it against your own event payloads in minutes.
Try the Live Demo View Pricing