szehon-ho commented on code in PR #60086: URL: https://github.com/apache/spark/pull/60086#discussion_r4472844197
########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,200 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it, producing the same history as if the events had arrived in order. A late insert or update that changes a tracked column adds an earlier version if it precedes all of the key's history, or closes the version it falls inside at its sequence value and opens a new one; a late delete closes the version it falls inside at the delete's sequence value. A delete that arrives before the inserts or updates it applies to is kept, and closes the version they open once they arrive. +- **Duplicate events** - A re-delivered event adds no version. + +For example, a first run receives two events for `id 1`: + +| id | name | version | op | +|----|--------|---------|--------| +| 1 | alice | 10 | UPSERT | +| 1 | alicia | 30 | UPSERT | + +The target holds two versions. The tables here omit the `__spark_autocdc_metadata` column (see [Auto CDC Considerations](#auto-cdc-considerations)): + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | alice | 10 | 10 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +A later run receives two late events, one inside the history and one before it: + +| id | name | version | op | +|----|--------|---------|--------| +| 1 | alyssa | 20 | UPSERT | +| 1 | ally | 5 | UPSERT | + +`alyssa` splits the `alice` version at 20, and `ally` adds a version before `alice`: + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | ally | 5 | 5 | 10 | +| 1 | alice | 10 | 10 | 20 | +| 1 | alyssa | 20 | 20 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +#### Defining a Type 2 Flow + +In Python, pass `stored_as_scd_type=2` to `create_auto_cdc_flow`: + +```python +from pyspark import pipelines as dp +from pyspark.sql import DataFrame + [email protected] +def cdc_events() -> DataFrame: + return spark.readStream.table("cdc_source") + +dp.create_streaming_table("customers") + +dp.create_auto_cdc_flow( + target="customers", + source="cdc_events", + keys=["id"], + sequence_by="version", + apply_as_deletes="op = 'DELETE'", + except_column_list=["op"], + stored_as_scd_type=2, +) +``` + +In SQL, add `STORED AS SCD TYPE 2`: + +```sql +CREATE STREAMING TABLE customers +FLOW AUTO CDC +FROM STREAM(cdc_events) +KEYS (id) +APPLY AS DELETE WHEN op = 'DELETE' +SEQUENCE BY version +COLUMNS * EXCEPT (op) +STORED AS SCD TYPE 2; +``` + +#### Choosing Which Columns Open a New Version + +By default, every column written to the target except the keys is tracked, including the sequencing column. Because every event for a key has its own sequence value, each insert or update then opens a new version. Review Comment: Could we clarify that, under default tracking, **distinct sequence values per key** create separate versions **when the sequencing column is retained in the target**? If that column is excluded, events with unchanged tracked values can coalesce. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
