Diving Deeper into Psyberg: Stateless vs Stateful Data Processing

Psyberg’s Two Processing Patterns

Psyberg, the incremental processing framework built by Netflix’s Membership and Finance data engineering team, operates in one of two modes. Choosing the right pattern depends on the nature of late-arriving data and how the target table’s columns are derived.

• Stateless Data Processing: Use this when the target table’s columns depend only on the content of individual incoming events, regardless of order. For example, a fact table tracking customer signups can simply append each new or late signup record as it arrives; each event is independent and has a single source.
• Stateful Data Processing: Use this when output depends on a sequence of events across one or more input streams. Tracking a customer account lifecycle—creation, upgrades, downgrades, cancellation—requires the correct sequence to derive attributes like account lifetime or current plan. A missed event leads to incorrect derived state. Late-arriving data in this mode must overwrite previously processed data to account for all events.

The Initialization Phase

Every Psyberg run begins with an initialization step that takes pipeline-specific parameters and computes the correct data range for processing.

Stateless Initialization

Consider a signup fact table running hourly, with raw signup events stored in an Iceberg table partitioned by landing date, hour, and batch ID. The YAML configuration below shows how etl_pattern_id=1 signals a stateless pattern.

- job:
   id: psyberg_session_init
   type: Spark
   spark:
     app_args:
       - --process_name=signup_fact_load
       - --src_tables=raw_signups
       - --psyberg_session_id=20230914061001
       - --psyberg_hwm_table=high_water_mark_table
       - --psyberg_session_table=psyberg_session_metadata
       - --etl_pattern_id=1

Psyberg uses the provided inputs to detect Iceberg snapshots created after the latest high watermark in the watermark table. From the summary column in snapshot metadata, it parses partition information for each snapshot. These processing URIs—JSON strings combining landing date, hour, and batch IDs—are stored in the psyberg_session_f table. The subsequent LOAD.FACT_TABLE job consumes this metadata for filtering and debugging.

Stateful Initialization

For stateful processing, the initialization differs. Building a cancellation fact table requires two inputs: raw cancellation events and a fact table of customer cancellation requests. These are combined to derive attributes like churn type (voluntary vs. involuntary).

- job:
   id: psyberg_session_init
   type: Spark
   spark:
     app_args:
       - --process_name=cancel_fact_load
       - --src_tables=raw_cancels|processing_ts,cancel_request_fact
       - --psyberg_session_id=20230914061501
       - --psyberg_hwm_table=high_water_mark_table
       - --psyberg_session_table=psyberg_session_metadata
       - --etl_pattern_id=2

Here, etl_pattern_id=2 marks the pipeline as stateful. Notice the processing_ts field on the raw_cancels source—this is the event processing timestamp, distinct from the Iceberg snapshot commit timestamp (event_landing_ts).

Capturing the consolidated range of events across all sources is critical, especially with late arrivals. Different timestamp fields can track source snapshot changes; using the wrong one risks missing events. Psyberg parses partition information from each source snapshot to determine actual partitions.

Psyberg queries the partitions metadata table for each source’s min and max column ranges. For the processing_ts column, the minimum value sets the lower limit of the data range—the derived minimum date and hour based on the input epoch timestamp.

Psyberg also tracks the Valid To TimeStamp (VTTS) of every input stream and computes the minimum across all of them. This establishes an upper processing limit, restricting the load to periods where all streams are complete.

From this metadata, Psyberg calculates minimum and maximum processing date/hour and event landing date/hour. These, along with other metadata, are persisted in psyberg_session_f.

Load, Audit, Publish

The ETL job uses the Write Audit Publish (WAP) pattern: it writes to an uncommitted Iceberg snapshot, audits it, then publishes. The LOAD.FACT_TABLE job takes psyberg_session_id and process_name as inputs.

For a stateless pattern, the job reads the processing URIs from psyberg_session_f and filters the source table accordingly. Late signup events are appended to the relevant target partitions. All writes stage in the uncommitted snapshot.

In stateful processing, the ETL reads the derived min/max date-hour range from psyberg_session_f and applies it as a filter across all input tables. After applying cancellation business logic, the job produces the cancel fact table with derived columns like cancellation type. Late-arriving events are handled by automatically extending the range, ensuring state changes are recomputed correctly.

Audits run directly on the uncommitted snapshot. Psyberg metadata identifies the exact cohort of data in the job run, which makes blocking checks like source-to-target count comparisons and missing-event detection efficient. Only after audits pass is the data published to the target table.

High Watermark Commit

Finally, Psyberg determines the latest timestamp associated with the Iceberg snapshot in the current run. This value updates the high watermark table, allowing the next pipeline instance to pick up new changes from that point onward. This completes the cycle, ready for the following run.