Skip to content

Commit 093c73a

Browse files
eric-thomas-dagstereric-thomas-dagster
committed
feat(dv2): Data Vault 2.0 Hub/Link/Satellite modeling component
Takes a raw source DataFrame and generates the Data Vault 2.0 modeling tables (Hub / Link / Satellite) with all the standard system columns — hash_key (SHA-256 or MD5 of business keys), hash_diff (SHA-256 of descriptive columns for change detection), load_date, record_source. Each output is a Dagster SDA — independently materializable, with real lineage back to the raw source asset. Right for the F500 POC "SDA vs task-scheduling" evaluation with a DV2.0-modeled warehouse layer. Emits 2 assets (hub + sat) by default, 3 (hub + link + sat) when link_business_keys is provided. Bumps version to 0.10.40.
1 parent ed65226 commit 093c73a

9 files changed

Lines changed: 260 additions & 2 deletions

File tree

‎assets/transforms/data_vault_hub_link_satellite/README.md‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
# DataVaultHubLinkSatelliteComponent
2+
3+
Take a raw source DataFrame and generate **Data Vault 2.0** Hub / Link / Satellite tables with all the standard system columns. Emits 2–3 assets per config:
4+
5+
- `<entity>_hub` — unique business keys + `hash_key` + `load_date` + `record_source`
6+
- `<entity>_sat` — descriptive attributes + `hash_key` + `hash_diff` + `load_date` + `record_source`
7+
- `<entity>_link` (optional) — combines hash keys of two-or-more hubs to represent a relationship
8+
9+
Each asset is **independently materializable** via SDA — you see full lineage back to the raw source, and you can partition / schedule / test each layer separately. This is the SDA-vs-task-scheduling story for DV2.0 modeling: instead of one big DAG task that populates raw → hub → sat → link, you get four Dagster assets with proper dependency + observability.
10+
11+
## Example — Customer entity
12+
13+
```yaml
14+
type: dagster_community_components.DataVaultHubLinkSatelliteComponent
15+
attributes:
16+
entity: customer
17+
upstream_asset_key: raw_customers
18+
business_keys: [customer_id]
19+
satellite_columns: [name, email, phone, address, updated_at]
20+
record_source: "erp_customers"
21+
group_name: data_vault_raw
22+
```
23+
24+
Emits `customer_hub` + `customer_sat`.
25+
26+
## Example — Customer↔Order link
27+
28+
```yaml
29+
attributes:
30+
entity: customer_order
31+
upstream_asset_key: raw_customer_orders
32+
business_keys: [customer_id, order_id]
33+
link_business_keys: [customer_id, order_id]
34+
satellite_columns: [order_status, order_amount, updated_at]
35+
record_source: "erp_orders"
36+
```
37+
38+
Emits `customer_order_hub` + `customer_order_link` + `customer_order_sat`.
39+
40+
## System columns
41+
42+
| Column | Type | Notes |
43+
|---|---|---|
44+
| `hash_key` | string | SHA-256 (default) or MD5 of the business-key columns |
45+
| `hash_diff` | string | Only on satellites — hash of descriptive columns, for change detection |
46+
| `load_date` | timestamp | UTC materialization timestamp |
47+
| `record_source` | string | Configurable source tag |
48+
49+
## Chaining
50+
51+
Common downstream shape:
52+
- Raw source ingestion (e.g. `mssql_workspace` or `oracle_ingestion`) → `data_vault_hub_link_satellite` → downstream marts via `dataframe_to_bigquery` / `warehouse_pipeline` / dbt.

‎assets/transforms/data_vault_hub_link_satellite/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
from .component import DataVaultHubLinkSatelliteComponent
2+
3+
__all__ = ["DataVaultHubLinkSatelliteComponent"]

‎assets/transforms/data_vault_hub_link_satellite/component.py‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
"""Data Vault 2.0 Hub / Link / Satellite Component.
2+
3+
Take a raw source DataFrame and generate the Data Vault 2.0 modeling
4+
tables — Hubs, Links, and Satellites — with all the required system
5+
columns (hash keys, load date, record source, hash diff). Emits three
6+
Dagster assets per config: `{prefix}_hub`, `{prefix}_link` (if link
7+
business keys are provided), and `{prefix}_sat`.
8+
9+
Right for teams migrating raw source ingestion into a Data Vault 2.0
10+
warehouse layer inside Dagster — SDA (Software-Defined Assets) lets
11+
you materialize hubs / links / satellites independently and see their
12+
lineage back to the raw source.
13+
"""
14+
from typing import List, Optional
15+
16+
import dagster as dg
17+
from pydantic import Field
18+
19+
20+
class DataVaultHubLinkSatelliteComponent(dg.Component, dg.Model, dg.Resolvable):
21+
"""Generate DV2.0 Hub, Link, and Satellite assets from a source DataFrame.
22+
23+
DV2.0 modeling primitives:
24+
- **Hub** = unique business keys + hash key + load-date + record-source
25+
- **Link** = combines hash keys of two-or-more hubs to represent a
26+
relationship (many-to-many)
27+
- **Satellite** = descriptive attributes hanging off a hub or link
28+
+ hash diff for change detection + effective-from date
29+
30+
Every table gets the standard system columns:
31+
- `hash_key` (SHA-256 of business-key columns)
32+
- `load_date` (materialization timestamp)
33+
- `record_source` (e.g. "orders_source")
34+
- Satellite only: `hash_diff` (SHA-256 of descriptive columns)
35+
36+
Example (Customer hub + descriptive satellite):
37+
38+
```yaml
39+
type: dagster_community_components.DataVaultHubLinkSatelliteComponent
40+
attributes:
41+
entity: customer
42+
upstream_asset_key: raw_customers
43+
business_keys: [customer_id]
44+
satellite_columns: [name, email, phone, address, updated_at]
45+
record_source: "erp_customers"
46+
group_name: data_vault_raw
47+
```
48+
49+
Example (Customer-to-Order link with sat on the link):
50+
51+
```yaml
52+
attributes:
53+
entity: customer_order
54+
upstream_asset_key: raw_customer_orders
55+
business_keys: [customer_id, order_id]
56+
link_business_keys: [customer_id, order_id] # keys that identify the link
57+
satellite_columns: [order_status, order_amount, updated_at]
58+
record_source: "erp_orders"
59+
```
60+
"""
61+
62+
entity: str = Field(description="Base name for emitted assets: <entity>_hub / _link / _sat.")
63+
upstream_asset_key: str = Field(description="Upstream DataFrame asset key to consume.")
64+
business_keys: List[str] = Field(description="Business key column(s) that identify the entity in the source.")
65+
satellite_columns: List[str] = Field(description="Descriptive columns that go into the satellite.")
66+
link_business_keys: Optional[List[str]] = Field(
67+
default=None,
68+
description=(
69+
"If provided, also emit a `<entity>_link` asset. These are the "
70+
"columns that jointly identify the LINK (usually the business "
71+
"keys of both endpoints)."
72+
),
73+
)
74+
75+
record_source: str = Field(description="Value populated in the record_source column.")
76+
load_date_column: str = Field(default="load_date", description="Name of the load-date system column.")
77+
hash_algo: str = Field(default="sha256", description="Hash algorithm: sha256 | md5")
78+
79+
group_name: Optional[str] = Field(default=None)
80+
asset_key_prefix: Optional[List[str]] = Field(default=None, description="Prefix for emitted asset keys.")
81+
82+
def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
83+
_self = self
84+
prefix = self.asset_key_prefix or []
85+
86+
def _hash_col(df, cols):
87+
import hashlib
88+
def _row(r):
89+
joined = "||".join(str(r[c]) if c in r.index else "" for c in cols)
90+
h = hashlib.sha256(joined.encode()) if _self.hash_algo == "sha256" else hashlib.md5(joined.encode())
91+
return h.hexdigest()
92+
return df.apply(_row, axis=1)
93+
94+
hub_key = dg.AssetKey([*prefix, f"{self.entity}_hub"])
95+
sat_key = dg.AssetKey([*prefix, f"{self.entity}_sat"])
96+
link_key = dg.AssetKey([*prefix, f"{self.entity}_link"]) if self.link_business_keys else None
97+
upstream = dg.AssetKey(self.upstream_asset_key)
98+
99+
@dg.asset(key=hub_key, group_name=self.group_name, compute_kind="data_vault",
100+
ins={"raw": dg.AssetIn(key=upstream)})
101+
def _hub(context: dg.AssetExecutionContext, raw):
102+
import pandas as pd
103+
df = raw.copy() if hasattr(raw, "copy") else pd.DataFrame(raw)
104+
hub = df[_self.business_keys].drop_duplicates().reset_index(drop=True)
105+
hub["hash_key"] = _hash_col(hub, _self.business_keys)
106+
hub[_self.load_date_column] = pd.Timestamp.utcnow()
107+
hub["record_source"] = _self.record_source
108+
context.add_output_metadata({"row_count": len(hub), "entity": _self.entity})
109+
return hub
110+
111+
@dg.asset(key=sat_key, group_name=self.group_name, compute_kind="data_vault",
112+
ins={"raw": dg.AssetIn(key=upstream)})
113+
def _sat(context: dg.AssetExecutionContext, raw):
114+
import pandas as pd
115+
df = raw.copy() if hasattr(raw, "copy") else pd.DataFrame(raw)
116+
keep_cols = _self.business_keys + [c for c in _self.satellite_columns if c in df.columns]
117+
sat = df[keep_cols].copy()
118+
sat["hash_key"] = _hash_col(df, _self.business_keys)
119+
sat["hash_diff"] = _hash_col(df, [c for c in _self.satellite_columns if c in df.columns])
120+
sat[_self.load_date_column] = pd.Timestamp.utcnow()
121+
sat["record_source"] = _self.record_source
122+
context.add_output_metadata({"row_count": len(sat), "entity": _self.entity})
123+
return sat
124+
125+
assets = [_hub, _sat]
126+
127+
if _self.link_business_keys:
128+
@dg.asset(key=link_key, group_name=self.group_name, compute_kind="data_vault",
129+
ins={"raw": dg.AssetIn(key=upstream)})
130+
def _link(context: dg.AssetExecutionContext, raw):
131+
import pandas as pd
132+
df = raw.copy() if hasattr(raw, "copy") else pd.DataFrame(raw)
133+
link = df[_self.link_business_keys].drop_duplicates().reset_index(drop=True)
134+
link["hash_key"] = _hash_col(link, _self.link_business_keys)
135+
# Also compute the referenced hub hash keys for lineage.
136+
for bk in _self.link_business_keys:
137+
link[f"{bk}_hash_key"] = _hash_col(link, [bk])
138+
link[_self.load_date_column] = pd.Timestamp.utcnow()
139+
link["record_source"] = _self.record_source
140+
context.add_output_metadata({"row_count": len(link), "entity": _self.entity})
141+
return link
142+
143+
assets.append(_link)
144+
145+
return dg.Definitions(assets=assets)

‎assets/transforms/data_vault_hub_link_satellite/example.yaml‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
type: dagster_community_components.DataVaultHubLinkSatelliteComponent
2+
attributes:
3+
entity: customer
4+
upstream_asset_key: raw_customers
5+
business_keys: [customer_id]
6+
satellite_columns: [name, email, phone, address, updated_at]
7+
record_source: "erp_customers"
8+
group_name: data_vault_raw
9+
10+
# Emit a link asset too (link between customer + order):
11+
# link_business_keys: [customer_id, order_id]
12+
13+
# asset_key_prefix: [dv, raw] # emitted keys become [dv, raw, customer_hub], etc

‎assets/transforms/data_vault_hub_link_satellite/requirements.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
pandas>=2.0.0

‎assets/transforms/data_vault_hub_link_satellite/schema.json‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
{
2+
"component_type": "dagster_community_components.DataVaultHubLinkSatelliteComponent",
3+
"description": "Take a raw source DataFrame and generate Data Vault 2.0 Hub / Link / Satellite tables with system columns (hash_key, load_date, record_source, hash_diff). Emits 2-3 assets per config: <entity>_hub, <entity>_sat, and optionally <entity>_link. Each asset is independently materializable via SDA — see lineage back to the raw source.",
4+
"category": "transformation",
5+
"icon": "AccountTree",
6+
"attributes": {
7+
"entity": {"type": "string", "label": "Entity Name", "required": true, "description": "Base name — emitted assets are <entity>_hub / _link / _sat."},
8+
"upstream_asset_key": {"type": "string", "label": "Upstream Asset Key", "required": true, "description": "Raw source DataFrame asset key to consume."},
9+
"business_keys": {"type": "array", "label": "Business Keys", "required": true, "description": "Business key column(s) that identify the entity in the source.", "x-dagster-widget": "columns"},
10+
"satellite_columns": {"type": "array", "label": "Satellite Columns", "required": true, "description": "Descriptive columns that go into the satellite.", "x-dagster-widget": "columns"},
11+
"link_business_keys": {"type": "array", "label": "Link Business Keys", "required": false, "description": "If provided, emit a <entity>_link asset. Usually business keys of both endpoints.", "x-dagster-widget": "columns"},
12+
"record_source": {"type": "string", "label": "Record Source", "required": true, "description": "Value populated in the record_source system column."},
13+
"load_date_column": {"type": "string", "label": "Load Date Column Name", "required": false, "default": "load_date"},
14+
"hash_algo": {"type": "string", "label": "Hash Algorithm", "required": false, "default": "sha256", "enum": ["sha256", "md5"], "ui:widget": "select"},
15+
"group_name": {"type": "string", "label": "Group Name", "required": false},
16+
"asset_key_prefix": {"type": "array", "label": "Asset Key Prefix", "required": false, "description": "Optional prefix segments prepended to every emitted asset key."}
17+
}
18+
}

‎dagster_community_components/__init__.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1047,6 +1047,7 @@
10471047
"MLflowWorkspaceComponent": "integrations/mlflow_workspace/component.py",
10481048
"WandbWorkspaceComponent": "integrations/wandb_workspace/component.py",
10491049
"PagerDutyWorkspaceComponent": "integrations/pagerduty_workspace/component.py",
1050+
"DataVaultHubLinkSatelliteComponent": "assets/transforms/data_vault_hub_link_satellite/component.py",
10501051
}
10511052

10521053
_loaded: dict[str, object] = {}

‎manifest.json‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33783,9 +33783,34 @@
3378333783
"requires_pip": ["pandas", "requests"],
3378433784
"chains_with": ["dataframe_to_snowflake", "pagerduty_incident_sensor"]
3378533785
}
33786+
},
33787+
{
33788+
"id": "data_vault_hub_link_satellite",
33789+
"name": "Data Vault 2.0 Hub / Link / Satellite",
33790+
"component_type": "dagster_community_components.DataVaultHubLinkSatelliteComponent",
33791+
"path": "assets/transforms/data_vault_hub_link_satellite",
33792+
"category": "transformation",
33793+
"icon": "AccountTree",
33794+
"description": "Take a raw source DataFrame and generate Data Vault 2.0 Hub / Link / Satellite tables with system columns (hash_key, load_date, record_source, hash_diff). Each is an independently-materializable SDA — see lineage back to the raw source.",
33795+
"tags": ["transformation", "data-vault", "dv2", "dv2.0", "modeling", "warehouse"],
33796+
"component_url": "https://raw.githubusercontent.com/eric-thomas-dagster/dagster-component-templates/main/assets/transforms/data_vault_hub_link_satellite/component.py",
33797+
"example_url": "https://raw.githubusercontent.com/eric-thomas-dagster/dagster-component-templates/main/assets/transforms/data_vault_hub_link_satellite/example.yaml",
33798+
"readme_url": "https://raw.githubusercontent.com/eric-thomas-dagster/dagster-component-templates/main/assets/transforms/data_vault_hub_link_satellite/README.md",
33799+
"requirements_url": "https://raw.githubusercontent.com/eric-thomas-dagster/dagster-component-templates/main/assets/transforms/data_vault_hub_link_satellite/requirements.txt",
33800+
"schema_url": "https://raw.githubusercontent.com/eric-thomas-dagster/dagster-component-templates/main/assets/transforms/data_vault_hub_link_satellite/schema.json",
33801+
"dependencies": {"pip": ["pandas>=2.0.0"]},
33802+
"validation": {"level": "code"},
33803+
"version": "1.0.0",
33804+
"author": "Dagster Community",
33805+
"agent_hints": {
33806+
"inputs": ["raw_source_dataframe_asset_key"],
33807+
"outputs": ["<entity>_hub", "<entity>_sat", "<entity>_link (optional)"],
33808+
"requires_pip": ["pandas"],
33809+
"chains_with": ["dataframe_to_bigquery", "dataframe_to_snowflake", "warehouse_pipeline"]
33810+
}
3378633811
}
3378733812
],
3378833813
"schema_spec": "https://raw.githubusercontent.com/eric-thomas-dagster/dagster-component-templates/main/schema-spec.json",
3378933814
"schema_spec_local": "schema-spec.json",
33790-
"total": 925
33815+
"total": 926
3379133816
}

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
44

55
[project]
66
name = "dagster-community-components"
7-
version = "0.10.39"
7+
version = "0.10.40"
88
description = "Community-maintained Dagster components — ingestion, transforms, IO managers, sensors, sinks, resources, and more."
99
readme = "README.md"
1010
requires-python = ">=3.10"

0 commit comments

Comments
 (0)