Enriching behavioral events before they become someone else's incident

Browser and mobile behavioral events are useful only after they are valid, complete, and consistent enough for analytics. Harvest is easy. The work is the path after harvest: store what actually arrived, process it in Flink, and publish only what a downstream team can query without inheriting an incident.

The figure above is that path. Collectors harvest events. Ingestion lands a raw copy and puts a stream on Kafka. Flink sanitizes, masks, windows, and deduplicates. Trusted records go to Snowflake and other serving stores. Rejects go to a bad-event Kafka topic, not into the analytical store of record.

Harvest

Harvest is collection, not enrichment. Browser and mobile SDKs emit clicks, page views, and app events. Collectors accept retries and late arrivals. The only contract that matters this early is event identity: a stable id (and usually a client timestamp) so later stages can tell a retry from a new fact.

Do not mask, join, or aggregate in the collector. If harvest “cleans” the payload, you can no longer reconstruct what the device sent when a dashboard is wrong.

Ingestion: raw store and Kafka

Ingestion has two jobs, and they are not the same topic.

Raw landing on S3 stores what arrived, as it arrived. That object is the harvest archive: partitioned by time, immutable, cheap to keep. When a Flink job is wrong, or a Kafka topic is retained shorter than an investigation, the raw landing is the replay source. It is not a query store.

Kafka ingestion topics are the live contract. Avro (or an equivalent schema) names fields, types, and required presence. Producers write; Flink reads. Partitioning should follow the key you will use for ordering and dedup — typically event id or a stable session/device key — so a restart does not reshuffle causality across tasks.

Kafka is not the archive. Retention is finite. If the only copy of a harvest day is a seven-day topic, you do not have ingestion. You have a buffer.

Processing: Flink

The enricher is a Flink job on those Kafka sources. It owns transformation: validate, sanitize, mask, window where the product needs time, deduplicate, shred into sink shapes, and split good records from rejects. It does not own harvest, and it does not own the Snowflake warehouse.

Sanitize and mask

Validation is fail-closed. A payload that cannot be interpreted — broken Avro, missing required fields, unknown event type — leaves on a side output toward the bad-event topic. It does not enter keyed state, windows, or Snowflake.

Sanitize is hygiene: trim, normalize enumerations, drop harvest-only headers. Mask is a privacy control. Fields that must not land in an analytical table (identifiers, free-text that can carry account data) are redacted or tokenized before the trusted sinks. Masking after Snowflake is an incident you scheduled yourself.

State: HashMap, RocksDB, and S3

Flink keyed state is how windows, dedup, and enrichment lookups survive a task failure. The backend is a product choice, not a default you forget to name.

HashMap state (heap) keeps state as Java objects. Latency is low. Cardinality is bounded by JVM heap. A full checkpoint copies that heap state. It is the right backend when the key space is small — a handful of enrichment maps, short windows, low-rate streams. It is the wrong backend for “every event id we have seen this week.” Heap will grow until the task manager dies, and the failure will look like Flink, not like a bad backend choice.

RocksDB is an embedded key-value store on local disk. State is serialized; hot data sits in block cache, the rest on disk. This is the backend for large keyed state: long windows, dedup sets, session state. Checkpoints can be incremental. The cost is serialization and I/O. If RocksDB local disks fill, or checkpointing to object storage stalls, the job back-pressures Kafka. Treat disk and checkpoint latency as capacity, the same way you treat Kafka lag.

S3 holds two different kinds of object, and mixing them is a design smell:

• the raw harvest landing from ingestion
• checkpoints and savepoints for the Flink job (RocksDB incremental files, metadata, savepoints for deploy and rewind)

A savepoint is an operator-aligned snapshot you take on purpose: upgrade the job, change parallelism, rewind. A checkpoint is the periodic snapshot Flink takes so a crash does not replay from the dawn of the topic. Both must restore operator state and Kafka offsets together. Restoring one without the other is how you double-count or skip.

Give operators UIDs. A topology change without UIDs drops state on the floor and your “exactly-once” story becomes a new consumer group.

Windowing

Windows are for questions that are about time, not for “we might see this event twice.”

Use event time for behavioral data. The client timestamp (bounded and watermarked) is the time of the click, not the time the Flink task processed it. Watermarks tell the window operator that no earlier event is expected. They are late by construction; that is the point.

• Tumbling windows for fixed buckets (minute, hour) into Snowflake grains.
• Sliding windows when overlapping intervals are the product.
• Session windows when a gap in activity closes a visit.

Allowed lateness keeps a window open for stragglers. Late events after that bound go to a side output — often the same operational path as other rejects — rather than silently mutating a closed bucket. Window state lives in HashMap or RocksDB like any other keyed state. A one-hour session window at high cardinality is a RocksDB decision whether anyone wrote it down or not.

Do not use a window as a substitute for dedup. A tumbling minute can still contain the same event id twice if the producer retried.

Offsets

Flink must own Kafka offsets. Disable consumer auto-commit. The committed position is stored in the checkpoint, next to window and dedup state.

On restore from checkpoint or savepoint, the job resumes those offsets and that state. If you resume offsets from an external consumer group but start with empty RocksDB, you will skip events whose state you no longer have, or you will apply windows to a stream as if you had never seen it. If you rewind Kafka without restoring state, you will emit duplicates into Snowflake.

Starting a new job is a different operation from restoring one. earliest, latest, and “whatever the group last committed” are three different historical claims. Write down which one you mean for first deploy, for poison-pill skip, and for a full replay from the S3 raw landing.

Checkpoint interval is a latency-versus-replay trade. Too rare, and a crash replays minutes of Kafka through dedup and windows. Too frequent, and S3 checkpoint I/O becomes the bottleneck. Alignment with Kafka transaction / idempotent sinks is what makes end-to-end exactly-once more than a slide.

Deduplication

Producer retries, collector retries, and Flink restarts all produce the same physical event more than once. Dedup is keyed state on a real event id, with a TTL that matches how long a duplicate can legally arrive — not “forever,” and not “until the next window fires.”

The first observation of an id is the fact. Later observations of the same id are dropped (or counted as duplicates on a metric, not as new rows). State backend follows cardinality: HashMap only if the id set fits in heap for the TTL; otherwise RocksDB.

Sinks still need their own idempotence. Snowflake should merge on that event id. Kafka sinks should use a deterministic key. Dedup in Flink plus a non-idempotent INSERT is how a replay weekend becomes someone else’s dashboard incident.

Downstream: Snowflake, bad-event Kafka, other stores

Snowflake is the trusted analytical sink: sanitized, masked, windowed as required, de-duplicated. It does not receive poison payloads.

Bad-event Kafka is the reject sink: schema failures, failed masks, events that cannot be keyed, and (if you choose) late-beyond-lateness. Operators replay from here after a contract fix — or from the S3 raw landing if the reject itself is incomplete. Do not load these rows into Snowflake “so we don’t lose data.” You would be preserving the incident.

Other stores — additional Kafka topics, DynamoDB, downstream APIs — get a shredded shape and an authorization model of their own. A successful Flink sink write is not proof that every consumer’s access policy was satisfied.

What I would inspect in a review

• Is there a raw landing that outlives Kafka retention?
• Does Flink checkpoint offsets with keyed state, and are auto-commits off?
• Is the state backend explicit (HashMap vs RocksDB) for the actual key cardinality?
• Are checkpoints and savepoints on object storage with operator UIDs?
• Is event-time watermarking defined, and is lateness a side output rather than a silent rewrite?
• Is dedup on event id with TTL, and are sinks idempotent on that id?
• Can a reject reach Snowflake? If yes, the split is not real.

Streaming code is the easy part once those answers are written down. The enricher’s job is to make a behavioral event into a trusted record — or hold it on the bad-event topic — before it becomes another team’s production issue.