From 283acf1a953a9ac72b928b5476c4ae96c25688fe Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Thu, 16 Jul 2026 09:58:50 +0100 Subject: [PATCH 01/10] add dataset builder docs --- .../create-training-datasets/index.md | 181 ++++++++++++++++++ docs/signals/applications/index.md | 1 + 2 files changed, 182 insertions(+) create mode 100644 docs/signals/applications/create-training-datasets/index.md diff --git a/docs/signals/applications/create-training-datasets/index.md b/docs/signals/applications/create-training-datasets/index.md new file mode 100644 index 000000000..73da07ad1 --- /dev/null +++ b/docs/signals/applications/create-training-datasets/index.md @@ -0,0 +1,181 @@ +--- +title: "Create training datasets from Signals attributes" +sidebar_position: 40 +sidebar_label: "Create training datasets" +description: "Build labeled training datasets from your Signals attributes using the Python SDK. Define anchor events, generate SQL, and execute against your warehouse." +keywords: ["training datasets", "machine learning", "dataset builder", "anchors", "signals python sdk", "snowflake"] +date: "2026-07-15" +--- + +Use the Signals Python SDK to build labeled training datasets from your [attribute groups](/docs/signals/attributes/attribute-groups/index.md). The dataset builder generates SQL that computes attribute values over historical data and joins them with labeled anchor events, producing a dataset ready for model training. + +Start by [connecting to Signals](/docs/signals/connection/index.md) to create a `Signals` client object. + +## How it works + +The dataset builder produces a training dataset in three stages: + +1. **Anchors** - identify moments in historical sessions where a user either achieved a goal (positive label) or did not (negative label) +2. **Attributes** - compute attribute values for each anchor event using only the events that preceded it in the session +3. **Assembly** - join anchors with their computed attribute values into a single labeled dataset + +This approach enforces point-in-time correctness: each row in the training dataset reflects only information that would have been available at the moment of the anchor event. This prevents data leakage, where a model trains on information it would not have at prediction time. + +You define what constitutes a positive outcome (the goal), the time window to scan, and which attribute groups provide the features. The SDK generates the SQL for all three stages, which you can inspect, save to disk, or execute directly against your warehouse. + +## Define anchors + +Anchors are the labeled events that form the rows of your training dataset. Each anchor is a point in a session that receives a label: `1` if the user achieved the goal, `0` if they did not. + +Use `SessionAnchors` when you want Signals to automatically identify anchor events from your raw event data. Use `UserSuppliedAnchors` when you already have a table of labeled events, for example from an existing ML pipeline or a manually curated dataset. + +### Session anchors + +Use `SessionAnchors` to automatically generate anchors from your event data. You specify a goal - the criteria that define a positive outcome - and a time window to scan. + +```python +from datetime import datetime, timezone + +from snowplow_signals import ( + Criteria, + Criterion, + EventProperty, + SessionAnchors, + TrainingSpan, +) + +anchors = SessionAnchors( + goal_criteria=Criteria( + any=[ + Criterion.eq( + EventProperty( + vendor="com.snowplowanalytics.snowplow.ecommerce", + name="snowplow_ecommerce_action", + major_version=1, + path="type", + ), + "transaction", + ), + ] + ), + training_span=TrainingSpan( + start_time=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_time=datetime(2024, 4, 1, tzinfo=timezone.utc), + ), +) +``` + +| Argument | Description | Type | Required? | +| --- | --- | --- | --- | +| `goal_criteria` | Criteria that define a positive anchor (label=1) | `Criteria` | ✅ | +| `training_span` | Time window to scan for anchor events | `TrainingSpan` | ✅ | +| `min_events` | Minimum prior in-session events before a point is eligible as an anchor | `int` | Default: `1` | +| `max_anchors_per_session` | Maximum anchor events per session. `None` for unlimited | `int` or `None` | Default: `None` | +| `max_negative_ratio` | Ratio of negative to positive anchors for downsampling | `float` | Default: `5.0` | +| `excluded_events` | Events to exclude from anchor generation | `list` | Default: page_ping events excluded | +| `output` | Override the output table location for anchors | `WarehouseTable` | Default: `None` | + +### User-supplied anchors + +If you already have a table of labeled anchor events, use `UserSuppliedAnchors` instead. + +```python +from snowplow_signals import UserSuppliedAnchors, WarehouseTable + +anchors = UserSuppliedAnchors( + source=WarehouseTable( + database="analytics", + schema="ml", + table="my_anchor_events", + ), + has_label=True, +) +``` + +| Argument | Description | Type | Required? | +| --- | --- | --- | --- | +| `source` | Table containing your pre-built anchor events | `WarehouseTable` | ✅ | +| `has_label` | Whether the source table contains a label column | `bool` | Default: `True` | + +## Build the dataset + +Call `build_dataset_sql()` on your `Signals` client to generate the SQL bundle. Pass the attribute groups that provide your features and the anchors you defined. + +```python +from snowplow_signals import Signals + +sp_signals = Signals( + api_url=os.environ["SIGNALS_DEPLOYED_URL"], + api_key_id=os.environ["CONSOLE_API_KEY_ID"], + api_key=os.environ["CONSOLE_API_KEY"], + org_id=os.environ["CONSOLE_ORG_ID"], +) + +bundle = sp_signals.build_dataset_sql( + attribute_groups=[my_attribute_group], + anchors=anchors, +) +``` + +| Argument | Description | Type | Required? | +| --- | --- | --- | --- | +| `attribute_groups` | Attribute groups that provide the feature columns | `list[AttributeGroup]` | ✅ | +| `anchors` | Anchor configuration (`SessionAnchors` or `UserSuppliedAnchors`) | `Anchors` | ✅ | +| `attributes_database` | Database for intermediate attribute tables | `str` | Default: warehouse default | +| `attributes_schema` | Schema for intermediate attribute tables | `str` | Default: warehouse default | +| `attributes_table_prefix` | Prefix for intermediate attribute table names | `str` | Default: `"signals_attributes"` | +| `dataset` | Override the output table location for the final dataset | `WarehouseTable` | Default: `None` | +| `max_lookback_days` | Override the lookback window in days | `int` | Default: derived from attribute periods | + +The returned `DatasetBundle` contains the generated SQL files and metadata. You can inspect the files, save them to disk, or execute them against your warehouse. + +### Save SQL to disk + +Use `save_to()` to write the generated SQL files, a manifest, and a README to a directory. + +```python +bundle.save_to("./dataset_output") +``` + +This creates individual SQL files for each stage (anchors, attributes, assembly), a `manifest.json` with input configuration and output table mappings, and a `README.md` documenting the execution order. + +## Execute against your warehouse + +Execute the generated SQL directly against your warehouse to produce the training dataset. + +### Snowflake + +```python +import snowflake.connector +from snowplow_signals.execution.snowflake import SnowflakeConnection + +sf_conn = SnowflakeConnection( + snowflake.connector.connect( + account=os.environ["SNOWFLAKE_ACCOUNT"], + user=os.environ["SNOWFLAKE_USER"], + warehouse=os.environ["SNOWFLAKE_WAREHOUSE"], + private_key=private_key_der, + ) +) + +result = bundle.execute(sf_conn) +``` + +The `execute()` method runs each stage in order: creating the anchors table, computing attribute tables, and assembling the final dataset. If a stage fails, it raises an `ExecutionError` with details of the failed stage and a list of completed stages. + +### Get results as a DataFrame + +After execution, convert the result to a pandas DataFrame for analysis or model training. + +```python +df = result.to_pandas() +print(f"Dataset: {len(df)} rows, {len(df.columns)} columns") +print(df.head()) +``` + +You can also inspect the execution stages: + +```python +for stage in result.stages: + print(f"[{stage.status}] {stage.stage}: {stage.table.table}") +``` diff --git a/docs/signals/applications/index.md b/docs/signals/applications/index.md index bd5abd531..a1c720a48 100644 --- a/docs/signals/applications/index.md +++ b/docs/signals/applications/index.md @@ -18,3 +18,4 @@ This section covers: * [Services](/docs/signals/applications/services/index.md): stable interfaces that group attribute group versions together for your applications, recommended for production use * [Retrieve attributes](/docs/signals/applications/retrieve-attributes/index.md): fetch calculated attribute values using a service or directly from an attribute group * [Subscribe to interventions](/docs/signals/applications/subscribe/index.md): receive intervention payloads and react to them in your application +* [Create training datasets](/docs/signals/applications/create-training-datasets/index.md): create training datasets from Signals attributes From 787b1f8bdea4e2f3c6c80f0e6580e5d1de9eb0ae Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Fri, 17 Jul 2026 16:37:51 +0100 Subject: [PATCH 02/10] pr comments --- .../create-training-datasets/index.md | 181 --------------- docs/signals/applications/index.md | 1 - docs/signals/ml-training-datasets/index.md | 215 ++++++++++++++++++ 3 files changed, 215 insertions(+), 182 deletions(-) delete mode 100644 docs/signals/applications/create-training-datasets/index.md create mode 100644 docs/signals/ml-training-datasets/index.md diff --git a/docs/signals/applications/create-training-datasets/index.md b/docs/signals/applications/create-training-datasets/index.md deleted file mode 100644 index 73da07ad1..000000000 --- a/docs/signals/applications/create-training-datasets/index.md +++ /dev/null @@ -1,181 +0,0 @@ ---- -title: "Create training datasets from Signals attributes" -sidebar_position: 40 -sidebar_label: "Create training datasets" -description: "Build labeled training datasets from your Signals attributes using the Python SDK. Define anchor events, generate SQL, and execute against your warehouse." -keywords: ["training datasets", "machine learning", "dataset builder", "anchors", "signals python sdk", "snowflake"] -date: "2026-07-15" ---- - -Use the Signals Python SDK to build labeled training datasets from your [attribute groups](/docs/signals/attributes/attribute-groups/index.md). The dataset builder generates SQL that computes attribute values over historical data and joins them with labeled anchor events, producing a dataset ready for model training. - -Start by [connecting to Signals](/docs/signals/connection/index.md) to create a `Signals` client object. - -## How it works - -The dataset builder produces a training dataset in three stages: - -1. **Anchors** - identify moments in historical sessions where a user either achieved a goal (positive label) or did not (negative label) -2. **Attributes** - compute attribute values for each anchor event using only the events that preceded it in the session -3. **Assembly** - join anchors with their computed attribute values into a single labeled dataset - -This approach enforces point-in-time correctness: each row in the training dataset reflects only information that would have been available at the moment of the anchor event. This prevents data leakage, where a model trains on information it would not have at prediction time. - -You define what constitutes a positive outcome (the goal), the time window to scan, and which attribute groups provide the features. The SDK generates the SQL for all three stages, which you can inspect, save to disk, or execute directly against your warehouse. - -## Define anchors - -Anchors are the labeled events that form the rows of your training dataset. Each anchor is a point in a session that receives a label: `1` if the user achieved the goal, `0` if they did not. - -Use `SessionAnchors` when you want Signals to automatically identify anchor events from your raw event data. Use `UserSuppliedAnchors` when you already have a table of labeled events, for example from an existing ML pipeline or a manually curated dataset. - -### Session anchors - -Use `SessionAnchors` to automatically generate anchors from your event data. You specify a goal - the criteria that define a positive outcome - and a time window to scan. - -```python -from datetime import datetime, timezone - -from snowplow_signals import ( - Criteria, - Criterion, - EventProperty, - SessionAnchors, - TrainingSpan, -) - -anchors = SessionAnchors( - goal_criteria=Criteria( - any=[ - Criterion.eq( - EventProperty( - vendor="com.snowplowanalytics.snowplow.ecommerce", - name="snowplow_ecommerce_action", - major_version=1, - path="type", - ), - "transaction", - ), - ] - ), - training_span=TrainingSpan( - start_time=datetime(2024, 1, 1, tzinfo=timezone.utc), - end_time=datetime(2024, 4, 1, tzinfo=timezone.utc), - ), -) -``` - -| Argument | Description | Type | Required? | -| --- | --- | --- | --- | -| `goal_criteria` | Criteria that define a positive anchor (label=1) | `Criteria` | ✅ | -| `training_span` | Time window to scan for anchor events | `TrainingSpan` | ✅ | -| `min_events` | Minimum prior in-session events before a point is eligible as an anchor | `int` | Default: `1` | -| `max_anchors_per_session` | Maximum anchor events per session. `None` for unlimited | `int` or `None` | Default: `None` | -| `max_negative_ratio` | Ratio of negative to positive anchors for downsampling | `float` | Default: `5.0` | -| `excluded_events` | Events to exclude from anchor generation | `list` | Default: page_ping events excluded | -| `output` | Override the output table location for anchors | `WarehouseTable` | Default: `None` | - -### User-supplied anchors - -If you already have a table of labeled anchor events, use `UserSuppliedAnchors` instead. - -```python -from snowplow_signals import UserSuppliedAnchors, WarehouseTable - -anchors = UserSuppliedAnchors( - source=WarehouseTable( - database="analytics", - schema="ml", - table="my_anchor_events", - ), - has_label=True, -) -``` - -| Argument | Description | Type | Required? | -| --- | --- | --- | --- | -| `source` | Table containing your pre-built anchor events | `WarehouseTable` | ✅ | -| `has_label` | Whether the source table contains a label column | `bool` | Default: `True` | - -## Build the dataset - -Call `build_dataset_sql()` on your `Signals` client to generate the SQL bundle. Pass the attribute groups that provide your features and the anchors you defined. - -```python -from snowplow_signals import Signals - -sp_signals = Signals( - api_url=os.environ["SIGNALS_DEPLOYED_URL"], - api_key_id=os.environ["CONSOLE_API_KEY_ID"], - api_key=os.environ["CONSOLE_API_KEY"], - org_id=os.environ["CONSOLE_ORG_ID"], -) - -bundle = sp_signals.build_dataset_sql( - attribute_groups=[my_attribute_group], - anchors=anchors, -) -``` - -| Argument | Description | Type | Required? | -| --- | --- | --- | --- | -| `attribute_groups` | Attribute groups that provide the feature columns | `list[AttributeGroup]` | ✅ | -| `anchors` | Anchor configuration (`SessionAnchors` or `UserSuppliedAnchors`) | `Anchors` | ✅ | -| `attributes_database` | Database for intermediate attribute tables | `str` | Default: warehouse default | -| `attributes_schema` | Schema for intermediate attribute tables | `str` | Default: warehouse default | -| `attributes_table_prefix` | Prefix for intermediate attribute table names | `str` | Default: `"signals_attributes"` | -| `dataset` | Override the output table location for the final dataset | `WarehouseTable` | Default: `None` | -| `max_lookback_days` | Override the lookback window in days | `int` | Default: derived from attribute periods | - -The returned `DatasetBundle` contains the generated SQL files and metadata. You can inspect the files, save them to disk, or execute them against your warehouse. - -### Save SQL to disk - -Use `save_to()` to write the generated SQL files, a manifest, and a README to a directory. - -```python -bundle.save_to("./dataset_output") -``` - -This creates individual SQL files for each stage (anchors, attributes, assembly), a `manifest.json` with input configuration and output table mappings, and a `README.md` documenting the execution order. - -## Execute against your warehouse - -Execute the generated SQL directly against your warehouse to produce the training dataset. - -### Snowflake - -```python -import snowflake.connector -from snowplow_signals.execution.snowflake import SnowflakeConnection - -sf_conn = SnowflakeConnection( - snowflake.connector.connect( - account=os.environ["SNOWFLAKE_ACCOUNT"], - user=os.environ["SNOWFLAKE_USER"], - warehouse=os.environ["SNOWFLAKE_WAREHOUSE"], - private_key=private_key_der, - ) -) - -result = bundle.execute(sf_conn) -``` - -The `execute()` method runs each stage in order: creating the anchors table, computing attribute tables, and assembling the final dataset. If a stage fails, it raises an `ExecutionError` with details of the failed stage and a list of completed stages. - -### Get results as a DataFrame - -After execution, convert the result to a pandas DataFrame for analysis or model training. - -```python -df = result.to_pandas() -print(f"Dataset: {len(df)} rows, {len(df.columns)} columns") -print(df.head()) -``` - -You can also inspect the execution stages: - -```python -for stage in result.stages: - print(f"[{stage.status}] {stage.stage}: {stage.table.table}") -``` diff --git a/docs/signals/applications/index.md b/docs/signals/applications/index.md index a1c720a48..bd5abd531 100644 --- a/docs/signals/applications/index.md +++ b/docs/signals/applications/index.md @@ -18,4 +18,3 @@ This section covers: * [Services](/docs/signals/applications/services/index.md): stable interfaces that group attribute group versions together for your applications, recommended for production use * [Retrieve attributes](/docs/signals/applications/retrieve-attributes/index.md): fetch calculated attribute values using a service or directly from an attribute group * [Subscribe to interventions](/docs/signals/applications/subscribe/index.md): receive intervention payloads and react to them in your application -* [Create training datasets](/docs/signals/applications/create-training-datasets/index.md): create training datasets from Signals attributes diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md new file mode 100644 index 000000000..80b55f55f --- /dev/null +++ b/docs/signals/ml-training-datasets/index.md @@ -0,0 +1,215 @@ +--- +title: "Create ML training datasets" +sidebar_position: 32 +sidebar_label: "ML training datasets" +description: "Build labeled, point-in-time correct training datasets for machine learning models using the Signals Python SDK. Define anchor events, generate SQL, and execute against your warehouse." +keywords: ["training datasets", "machine learning", "dataset builder", "anchors", "signals python sdk", "snowflake", "ML"] +date: "2026-07-15" +--- + +Training a machine learning model on behavioral data requires a labeled dataset where each row represents a moment in time, with features computed only from events that had occurred up to that point. Building these datasets manually is error-prone: it is easy to accidentally include future information (data leakage), which produces models that perform well in testing but fail in production. + +The Signals dataset builder automates this process. You define the outcome you want to predict, and Signals generates SQL that scans your historical event data, identifies labeled anchor points, computes attribute values using only prior events, and assembles a ready-to-use training dataset. Because the dataset builder uses the same [attribute groups](/docs/signals/attributes/attribute-groups/index.md) that power your real-time Signals deployment, the features your model trains on are identical to the features it receives at inference time. + +Start by [connecting to Signals](/docs/signals/connection/index.md) to create a `Signals` client object. + +## Workflow + +Building and using an ML training dataset follows this process: + +1. [Define your attribute groups](/docs/signals/attributes/attribute-groups/index.md) with the features you want your model to learn from (e.g. `product_view_count`, `add_to_cart_count`) +2. Define anchors that specify what outcome you want to predict and over what time window +3. Generate the SQL bundle using the dataset builder +4. Execute the SQL against your warehouse to produce the training dataset +5. Train your model on the resulting DataFrame +6. Deploy your model, serving it the same attributes in real time via [Retrieve attributes](/docs/signals/applications/retrieve-attributes/index.md) + +## How it works + +The dataset builder produces a training dataset in three stages: + +1. Anchors: scan sessions within the training window and identify anchor points. Sessions where the goal event occurred produce positive anchors (label=1). Sessions without the goal produce negative anchors (label=0). Negative anchors are downsampled to avoid class imbalance. +2. Attributes: for each anchor, compute attribute values using only the events that preceded the anchor timestamp in that session. This enforces point-in-time correctness, so attributes reflect only what was known at the moment of the anchor. +3. Assembly: join anchors with their computed attribute values into a single labeled dataset, with one row per anchor and one column per attribute. + +For example, if you define two attributes (`product_view_count` and `add_to_cart_count`) with a transaction goal, the final dataset looks like this: + +| domain_sessionid | anchor_ts | label | product_view_count | add_to_cart_count | +| --- | --- | --- | --- | --- | +| `abc-123` | 2024-01-15 09:32:00 | 1 | 5 | 2 | +| `def-456` | 2024-01-15 10:01:00 | 0 | 3 | 0 | +| `ghi-789` | 2024-01-16 14:22:00 | 1 | 8 | 4 | + +Each row captures the attribute values as they were at the anchor timestamp, not at the end of the session. A model trained on this data learns the same signal it will see when serving predictions in real time. + +## Define anchors + +Anchors are the labeled events that form the rows of your training dataset. Each anchor is a point in a session that receives a label: `1` if the user achieved the goal, `0` if they did not. + +Use `SessionAnchors` when you want Signals to automatically identify anchor events from your raw event data. Use `UserSuppliedAnchors` when you already have a table of labeled events, for example from an existing ML pipeline or a manually curated dataset. + +### Session anchors + +Use `SessionAnchors` to automatically generate anchors from your event data. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window, labels each based on whether the goal was achieved, and produces one anchor per qualifying session. + +```python +from datetime import datetime, timezone + +from snowplow_signals import ( + Criteria, + Criterion, + EventProperty, + SessionAnchors, + TrainingSpan, +) + +anchors = SessionAnchors( + goal_criteria=Criteria( + any=[ + Criterion.eq( + EventProperty( + vendor="com.snowplowanalytics.snowplow.ecommerce", + name="snowplow_ecommerce_action", + major_version=1, + path="type", + ), + "transaction", + ), + ] + ), + training_span=TrainingSpan( + start_time=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_time=datetime(2024, 4, 1, tzinfo=timezone.utc), + ), +) +``` + +| Argument | Description | Type | Required? | +| --- | --- | --- | --- | +| `goal_criteria` | Criteria that define a positive anchor (label=1) | `Criteria` | ✅ | +| `training_span` | Time window to scan for anchor events | `TrainingSpan` | ✅ | +| `min_events` | Minimum number of prior in-session events before an anchor is eligible. Increase this to ensure each anchor has enough behavioral signal for meaningful features. | `int` | Default: `1` | +| `max_anchors_per_session` | Maximum anchor events per session. `None` for unlimited. Set this to limit overrepresentation of long sessions. | `int` or `None` | Default: `None` | +| `max_negative_ratio` | Maximum ratio of negative to positive anchors. Negative anchors are downsampled to this ratio. Lower values produce more balanced datasets; higher values preserve more data. | `float` | Default: `5.0` | +| `excluded_events` | Events to exclude from anchor generation. By default, `page_ping` events are excluded because they do not represent meaningful user actions. | `list` | Default: `page_ping` events excluded | +| `output` | Override the output table location for the anchors table | `WarehouseTable` | Default: `None` | + +### User-supplied anchors + +If you already have a table of labeled anchor events, use `UserSuppliedAnchors` instead. Your table must contain the following columns: + +| Column | Type | Description | +| --- | --- | --- | +| Attribute key column (e.g. `domain_sessionid`) | `VARCHAR` | The attribute key used by your attribute groups. The column name must match the attribute key name. | +| `anchor_ts` | `TIMESTAMP` | The timestamp of the anchor event | +| `label` | `INTEGER` | `1` for positive, `0` for negative. Only required when `has_label` is `True`. | + +For example, if your attribute groups use `domain_sessionid` as the attribute key: + +| domain_sessionid | anchor_ts | label | +| --- | --- | --- | +| `abc-123` | 2024-01-15 09:32:00 | 1 | +| `def-456` | 2024-01-15 10:01:00 | 0 | + +```python +from snowplow_signals import UserSuppliedAnchors, WarehouseTable + +anchors = UserSuppliedAnchors( + source=WarehouseTable( + database="analytics", + schema="ml", + table="my_anchor_events", + ), + has_label=True, +) +``` + +| Argument | Description | Type | Required? | +| --- | --- | --- | --- | +| `source` | Table containing your pre-built anchor events | `WarehouseTable` | ✅ | +| `has_label` | Whether the source table contains a `label` column. Set to `False` if your anchors are unlabeled. | `bool` | Default: `True` | + +## Generate the SQL + +The `build_dataset_sql()` method sends your attribute group definitions and anchor configuration to the Signals API, which returns a bundle of SQL files. This is a separate step from execution so you can review the generated SQL, save it to version control, or run it in a different environment. + +```python +bundle = sp_signals.build_dataset_sql( + attribute_groups=[my_attribute_group], + anchors=anchors, +) +``` + +| Argument | Description | Type | Required? | +| --- | --- | --- | --- | +| `attribute_groups` | Attribute groups that provide the feature columns. Each attribute in these groups becomes a column in the final dataset. | `list[AttributeGroup]` | ✅ | +| `anchors` | Anchor configuration (`SessionAnchors` or `UserSuppliedAnchors`) | `Anchors` | ✅ | +| `attributes_database` | Database for the intermediate per-attribute-key tables created during execution | `str` | Default: warehouse default | +| `attributes_schema` | Schema for the intermediate per-attribute-key tables created during execution | `str` | Default: warehouse default | +| `attributes_table_prefix` | Prefix for intermediate table names. For example, with the default prefix and an attribute key of `domain_sessionid`, the table is named `signals_attributes_domain_sessionid`. | `str` | Default: `"signals_attributes"` | +| `dataset` | Override the output table location for the final assembled dataset | `WarehouseTable` | Default: `None` | +| `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. Override it to widen or narrow the event window. | `int` | Default: derived from attribute periods | + +The returned `DatasetBundle` contains the generated SQL files. You can inspect them, save them to disk, or execute them directly. + +### Save SQL to disk + +Use `save_to()` to write the generated SQL files, a manifest, and a README to a directory. This is useful for reviewing the SQL before execution or committing it to version control. + +```python +bundle.save_to("./dataset_output") +``` + +This creates: + +- Individual SQL files for each stage (`signals_anchors.sql`, `signals_attributes_domain_sessionid.sql`, `signals_training_dataset.sql`) +- `manifest.json` with input configuration and output table mappings +- `README.md` documenting the execution order + +## Execute against your warehouse + +Once you have a `DatasetBundle`, execute the SQL against your warehouse to produce the training dataset. The warehouse connection is separate from your Signals API credentials because the SQL runs directly against your data warehouse, not through the Signals API. + +### Snowflake + +Snowflake supports key-pair authentication for programmatic access. Wrap your Snowflake connection in a `SnowflakeConnection` and pass it to `execute()`. + +```python +import snowflake.connector +from snowplow_signals.execution.snowflake import SnowflakeConnection + +sf_conn = SnowflakeConnection( + snowflake.connector.connect( + account=os.environ["SNOWFLAKE_ACCOUNT"], + user=os.environ["SNOWFLAKE_USER"], + warehouse=os.environ["SNOWFLAKE_WAREHOUSE"], + private_key=private_key_der, + ) +) + +result = bundle.execute(sf_conn) +``` + +The `execute()` method runs three stages in order: + +1. Creates the anchors table (`signals_anchors`) +2. Creates one attribute table per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values +3. Assembles the final training dataset (`signals_training_dataset`) by joining all tables together + +If a stage fails, it raises an `ExecutionError` with details of the failed stage and a list of completed stages. + +### Get results as a DataFrame + +After execution, call `to_pandas()` to get the training dataset as a DataFrame, ready for model training. + +```python +df = result.to_pandas() +``` + +The resulting DataFrame contains one row per anchor, with columns for the attribute key, anchor timestamp, label, and every attribute from your attribute groups: + +| | domain_sessionid | anchor_ts | label | product_view_count | add_to_cart_count | +| --- | --- | --- | --- | --- | --- | +| 0 | abc-123 | 2024-01-15 09:32:00 | 1 | 5 | 2 | +| 1 | def-456 | 2024-01-15 10:01:00 | 0 | 3 | 0 | +| 2 | ghi-789 | 2024-01-16 14:22:00 | 1 | 8 | 4 | From 8861fb247e3225b8a0beaef8d39d3b90f0e45e85 Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Fri, 17 Jul 2026 16:53:31 +0100 Subject: [PATCH 03/10] reflect latest sdk changes --- docs/signals/ml-training-datasets/index.md | 66 +++++++++------------- 1 file changed, 28 insertions(+), 38 deletions(-) diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md index 80b55f55f..050950d8c 100644 --- a/docs/signals/ml-training-datasets/index.md +++ b/docs/signals/ml-training-datasets/index.md @@ -18,8 +18,8 @@ Start by [connecting to Signals](/docs/signals/connection/index.md) to create a Building and using an ML training dataset follows this process: 1. [Define your attribute groups](/docs/signals/attributes/attribute-groups/index.md) with the features you want your model to learn from (e.g. `product_view_count`, `add_to_cart_count`) -2. Define anchors that specify what outcome you want to predict and over what time window -3. Generate the SQL bundle using the dataset builder +2. Build the dataset SQL bundle by calling the appropriate method - `build_dataset_with_session_anchors()` for auto-generated anchors, or `build_dataset_with_custom_anchors()` for a pre-existing anchor table +3. Optionally inspect or save the generated SQL 4. Execute the SQL against your warehouse to produce the training dataset 5. Train your model on the resulting DataFrame 6. Deploy your model, serving it the same attributes in real time via [Retrieve attributes](/docs/signals/applications/retrieve-attributes/index.md) @@ -46,11 +46,11 @@ Each row captures the attribute values as they were at the anchor timestamp, not Anchors are the labeled events that form the rows of your training dataset. Each anchor is a point in a session that receives a label: `1` if the user achieved the goal, `0` if they did not. -Use `SessionAnchors` when you want Signals to automatically identify anchor events from your raw event data. Use `UserSuppliedAnchors` when you already have a table of labeled events, for example from an existing ML pipeline or a manually curated dataset. +Signals provides two approaches: session anchors (automatically derived from your event data) and user-supplied anchors (from a table you provide). You choose between them by calling different methods on the Signals client. ### Session anchors -Use `SessionAnchors` to automatically generate anchors from your event data. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window, labels each based on whether the goal was achieved, and produces one anchor per qualifying session. +Use `build_dataset_with_session_anchors()` to automatically generate anchors from your event data. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window, labels each based on whether the goal was achieved, and produces one anchor per qualifying session. ```python from datetime import datetime, timezone @@ -59,11 +59,11 @@ from snowplow_signals import ( Criteria, Criterion, EventProperty, - SessionAnchors, TrainingSpan, ) -anchors = SessionAnchors( +bundle = sp_signals.build_dataset_with_session_anchors( + attribute_groups=[my_attribute_group], goal_criteria=Criteria( any=[ Criterion.eq( @@ -86,23 +86,27 @@ anchors = SessionAnchors( | Argument | Description | Type | Required? | | --- | --- | --- | --- | +| `attribute_groups` | Attribute groups that provide the feature columns. Each attribute in these groups becomes a column in the final dataset. | `list[AttributeGroup]` | ✅ | | `goal_criteria` | Criteria that define a positive anchor (label=1) | `Criteria` | ✅ | | `training_span` | Time window to scan for anchor events | `TrainingSpan` | ✅ | | `min_events` | Minimum number of prior in-session events before an anchor is eligible. Increase this to ensure each anchor has enough behavioral signal for meaningful features. | `int` | Default: `1` | | `max_anchors_per_session` | Maximum anchor events per session. `None` for unlimited. Set this to limit overrepresentation of long sessions. | `int` or `None` | Default: `None` | | `max_negative_ratio` | Maximum ratio of negative to positive anchors. Negative anchors are downsampled to this ratio. Lower values produce more balanced datasets; higher values preserve more data. | `float` | Default: `5.0` | | `excluded_events` | Events to exclude from anchor generation. By default, `page_ping` events are excluded because they do not represent meaningful user actions. | `list` | Default: `page_ping` events excluded | -| `output` | Override the output table location for the anchors table | `WarehouseTable` | Default: `None` | +| `anchors_table` | Override the output table location for the anchors table | `WarehouseTable` | Default: `None` | +| `attributes_table` | Override the output table configuration for intermediate per-attribute-key tables (database, schema, table prefix) | `AttributesWarehouseTable` | Default: `None` | +| `dataset_table` | Override the output table location for the final assembled dataset | `WarehouseTable` | Default: `None` | +| `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. Override it to widen or narrow the event window. | `int` | Default: derived from attribute periods | ### User-supplied anchors -If you already have a table of labeled anchor events, use `UserSuppliedAnchors` instead. Your table must contain the following columns: +If you already have a table of labeled anchor events, use `build_dataset_with_custom_anchors()` instead. Your table must contain the following columns: | Column | Type | Description | | --- | --- | --- | | Attribute key column (e.g. `domain_sessionid`) | `VARCHAR` | The attribute key used by your attribute groups. The column name must match the attribute key name. | | `anchor_ts` | `TIMESTAMP` | The timestamp of the anchor event | -| `label` | `INTEGER` | `1` for positive, `0` for negative. Only required when `has_label` is `True`. | +| `label` | `INTEGER` | `1` for positive, `0` for negative. Only required when `anchors_have_label` is `True`. | For example, if your attribute groups use `domain_sessionid` as the attribute key: @@ -112,45 +116,31 @@ For example, if your attribute groups use `domain_sessionid` as the attribute ke | `def-456` | 2024-01-15 10:01:00 | 0 | ```python -from snowplow_signals import UserSuppliedAnchors, WarehouseTable +from snowplow_signals import WarehouseTable -anchors = UserSuppliedAnchors( - source=WarehouseTable( +bundle = sp_signals.build_dataset_with_custom_anchors( + attribute_groups=[my_attribute_group], + anchors_table=WarehouseTable( database="analytics", schema="ml", table="my_anchor_events", ), - has_label=True, -) -``` - -| Argument | Description | Type | Required? | -| --- | --- | --- | --- | -| `source` | Table containing your pre-built anchor events | `WarehouseTable` | ✅ | -| `has_label` | Whether the source table contains a `label` column. Set to `False` if your anchors are unlabeled. | `bool` | Default: `True` | - -## Generate the SQL - -The `build_dataset_sql()` method sends your attribute group definitions and anchor configuration to the Signals API, which returns a bundle of SQL files. This is a separate step from execution so you can review the generated SQL, save it to version control, or run it in a different environment. - -```python -bundle = sp_signals.build_dataset_sql( - attribute_groups=[my_attribute_group], - anchors=anchors, + anchors_have_label=True, ) ``` | Argument | Description | Type | Required? | | --- | --- | --- | --- | | `attribute_groups` | Attribute groups that provide the feature columns. Each attribute in these groups becomes a column in the final dataset. | `list[AttributeGroup]` | ✅ | -| `anchors` | Anchor configuration (`SessionAnchors` or `UserSuppliedAnchors`) | `Anchors` | ✅ | -| `attributes_database` | Database for the intermediate per-attribute-key tables created during execution | `str` | Default: warehouse default | -| `attributes_schema` | Schema for the intermediate per-attribute-key tables created during execution | `str` | Default: warehouse default | -| `attributes_table_prefix` | Prefix for intermediate table names. For example, with the default prefix and an attribute key of `domain_sessionid`, the table is named `signals_attributes_domain_sessionid`. | `str` | Default: `"signals_attributes"` | -| `dataset` | Override the output table location for the final assembled dataset | `WarehouseTable` | Default: `None` | -| `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. Override it to widen or narrow the event window. | `int` | Default: derived from attribute periods | +| `anchors_table` | Table containing your pre-built anchor events | `WarehouseTable` | ✅ | +| `anchors_have_label` | Whether the source table contains a `label` column. Set to `False` if your anchors are unlabeled. | `bool` | Default: `True` | +| `attributes_table` | Override the output table configuration for intermediate per-attribute-key tables (database, schema, table prefix) | `AttributesWarehouseTable` | Default: `None` | +| `dataset_table` | Override the output table location for the final assembled dataset | `WarehouseTable` | Default: `None` | +| `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. | `int` | Default: derived from attribute periods | + +## Inspect and save the SQL bundle -The returned `DatasetBundle` contains the generated SQL files. You can inspect them, save them to disk, or execute them directly. +Both `build_dataset_with_session_anchors()` and `build_dataset_with_custom_anchors()` return a `DatasetBundle` containing the generated SQL files. You can inspect them, save them to disk, or execute them directly. ### Save SQL to disk @@ -200,10 +190,10 @@ If a stage fails, it raises an `ExecutionError` with details of the failed stage ### Get results as a DataFrame -After execution, call `to_pandas()` to get the training dataset as a DataFrame, ready for model training. +After execution, access the `dataframe` attribute to get the training dataset as a pandas DataFrame, ready for model training. ```python -df = result.to_pandas() +df = result.dataframe ``` The resulting DataFrame contains one row per anchor, with columns for the attribute key, anchor timestamp, label, and every attribute from your attribute groups: From 8a3524b1288348c2b5265dea5858486c1b600825 Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Fri, 24 Jul 2026 10:01:27 +0100 Subject: [PATCH 04/10] update to final sdk --- docs/signals/ml-training-datasets/index.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md index 050950d8c..2912a642a 100644 --- a/docs/signals/ml-training-datasets/index.md +++ b/docs/signals/ml-training-datasets/index.md @@ -186,14 +186,14 @@ The `execute()` method runs three stages in order: 2. Creates one attribute table per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values 3. Assembles the final training dataset (`signals_training_dataset`) by joining all tables together -If a stage fails, it raises an `ExecutionError` with details of the failed stage and a list of completed stages. +If a stage fails, it raises an `ExecutionError` with the name of the failed stage, the target table, and the underlying cause. ### Get results as a DataFrame -After execution, access the `dataframe` attribute to get the training dataset as a pandas DataFrame, ready for model training. +After execution, call `to_pandas()` to pull the training dataset into a pandas DataFrame, ready for model training. By default, it fetches up to 10,000 rows. ```python -df = result.dataframe +df = result.to_pandas() ``` The resulting DataFrame contains one row per anchor, with columns for the attribute key, anchor timestamp, label, and every attribute from your attribute groups: From 2bb46cb7e3ec1caaeeba16fb7b115477bdf3dca8 Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Fri, 24 Jul 2026 12:45:51 +0100 Subject: [PATCH 05/10] pr comments --- docs/signals/ml-training-datasets/index.md | 62 ++++++++++++++++------ 1 file changed, 47 insertions(+), 15 deletions(-) diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md index 2912a642a..2008272db 100644 --- a/docs/signals/ml-training-datasets/index.md +++ b/docs/signals/ml-training-datasets/index.md @@ -7,12 +7,18 @@ keywords: ["training datasets", "machine learning", "dataset builder", "anchors" date: "2026-07-15" --- -Training a machine learning model on behavioral data requires a labeled dataset where each row represents a moment in time, with features computed only from events that had occurred up to that point. Building these datasets manually is error-prone: it is easy to accidentally include future information (data leakage), which produces models that perform well in testing but fail in production. +The Signals dataset builder lets you create labeled training datasets for machine learning models directly from your Snowplow event data. It generates point-in-time correct features from your existing [attribute groups](/docs/signals/attributes/attribute-groups/index.md), so the features your model trains on are identical to the features it receives at inference time. -The Signals dataset builder automates this process. You define the outcome you want to predict, and Signals generates SQL that scans your historical event data, identifies labeled anchor points, computes attribute values using only prior events, and assembles a ready-to-use training dataset. Because the dataset builder uses the same [attribute groups](/docs/signals/attributes/attribute-groups/index.md) that power your real-time Signals deployment, the features your model trains on are identical to the features it receives at inference time. +Training a machine learning model on behavioral data requires a dataset where each row represents a moment in time, with features computed only from events that had occurred up to that point. Building these datasets manually is error-prone: it is easy to accidentally include future information (data leakage), which produces models that perform well in testing but fail in production. The dataset builder automates this process, enforcing point-in-time correctness by construction. Start by [connecting to Signals](/docs/signals/connection/index.md) to create a `Signals` client object. +## Why point-in-time correctness matters + +When you serve predictions in real time, your model only has access to events that have happened so far in the session. If your training data includes attributes computed from the full session (including events after the prediction point), your model learns patterns it will never see in production. This is called data leakage, and it is the most common reason ML models degrade after deployment. + +The dataset builder prevents this by computing each attribute value using only events that occurred before the anchor timestamp. Because it uses the same attribute group definitions that power your real-time Signals deployment, training and serving are guaranteed to use the same feature logic. + ## Workflow Building and using an ML training dataset follows this process: @@ -28,6 +34,30 @@ Building and using an ML training dataset follows this process: The dataset builder produces a training dataset in three stages: +```mermaid +flowchart LR + subgraph stage1["1. Anchors"] + direction TB + A1["Scan sessions in\ntraining window"] --> A2{"Goal event\noccurred?"} + A2 -- Yes --> A3["Positive anchor\n(label = 1)"] + A2 -- No --> A4["Negative anchor\n(label = 0)"] + A3 --> A5["Downsample\nnegatives"] + A4 --> A5 + end + + subgraph stage2["2. Attributes"] + direction TB + B1["For each anchor,\nfind preceding events"] --> B2["Compute attribute\nvalues using only\nevents before\nanchor timestamp"] + end + + subgraph stage3["3. Assembly"] + direction TB + C1["Join anchors +\nattributes into\nlabeled dataset"] --> C2["One row per anchor\nOne column per attribute"] + end + + stage1 --> stage2 --> stage3 +``` + 1. Anchors: scan sessions within the training window and identify anchor points. Sessions where the goal event occurred produce positive anchors (label=1). Sessions without the goal produce negative anchors (label=0). Negative anchors are downsampled to avoid class imbalance. 2. Attributes: for each anchor, compute attribute values using only the events that preceded the anchor timestamp in that session. This enforces point-in-time correctness, so attributes reflect only what was known at the moment of the anchor. 3. Assembly: join anchors with their computed attribute values into a single labeled dataset, with one row per anchor and one column per attribute. @@ -50,7 +80,7 @@ Signals provides two approaches: session anchors (automatically derived from you ### Session anchors -Use `build_dataset_with_session_anchors()` to automatically generate anchors from your event data. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window, labels each based on whether the goal was achieved, and produces one anchor per qualifying session. +Use `build_dataset_with_session_anchors()` to automatically generate anchors from your event data. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window and labels each based on whether the goal was achieved. For positive sessions, anchors are placed at the goal event itself. For negative sessions (where the goal was never achieved), anchors are placed at randomly selected events within the session. Negative anchors are then downsampled according to `max_negative_ratio` to avoid class imbalance. ```python from datetime import datetime, timezone @@ -89,14 +119,14 @@ bundle = sp_signals.build_dataset_with_session_anchors( | `attribute_groups` | Attribute groups that provide the feature columns. Each attribute in these groups becomes a column in the final dataset. | `list[AttributeGroup]` | ✅ | | `goal_criteria` | Criteria that define a positive anchor (label=1) | `Criteria` | ✅ | | `training_span` | Time window to scan for anchor events | `TrainingSpan` | ✅ | -| `min_events` | Minimum number of prior in-session events before an anchor is eligible. Increase this to ensure each anchor has enough behavioral signal for meaningful features. | `int` | Default: `1` | -| `max_anchors_per_session` | Maximum anchor events per session. `None` for unlimited. Set this to limit overrepresentation of long sessions. | `int` or `None` | Default: `None` | -| `max_negative_ratio` | Maximum ratio of negative to positive anchors. Negative anchors are downsampled to this ratio. Lower values produce more balanced datasets; higher values preserve more data. | `float` | Default: `5.0` | +| `min_events` | Minimum number of prior in-session events before an anchor is eligible. Increase this to filter out anchors with too little behavioral signal, for example set to `5` to ensure each anchor has at least 5 prior events. | `int` | Default: `1` | +| `max_anchors_per_session` | Maximum anchor events per session. `None` for unlimited. Set this to limit overrepresentation of long sessions, for example set to `1` to ensure each session contributes at most one training example. | `int` or `None` | Default: `None` | +| `max_negative_ratio` | Maximum ratio of negative to positive anchors. Negative anchors are downsampled to this ratio. Lower values produce more balanced datasets; higher values preserve more data. For example, set to `1.0` for a balanced 1:1 dataset. | `float` | Default: `5.0` | | `excluded_events` | Events to exclude from anchor generation. By default, `page_ping` events are excluded because they do not represent meaningful user actions. | `list` | Default: `page_ping` events excluded | -| `anchors_table` | Override the output table location for the anchors table | `WarehouseTable` | Default: `None` | -| `attributes_table` | Override the output table configuration for intermediate per-attribute-key tables (database, schema, table prefix) | `AttributesWarehouseTable` | Default: `None` | -| `dataset_table` | Override the output table location for the final assembled dataset | `WarehouseTable` | Default: `None` | -| `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. Override it to widen or narrow the event window. | `int` | Default: derived from attribute periods | +| `anchors_table` | Custom output table location for the anchors table. By default, a table named `signals_anchors` is created in your warehouse. | `WarehouseTable` | Default: `None` | +| `attributes_table` | Override the output table location for the intermediate attribute tables. During execution, one table is created per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values. | `AttributesWarehouseTable` | Default: `None` | +| `dataset_table` | Custom output table location for the final assembled dataset. By default, a table named `signals_training_dataset` is created in your warehouse. | `WarehouseTable` | Default: `None` | +| `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. Set a lower value to narrow the event window, or a higher value to include older events. | `int` | Default: derived from attribute periods | ### User-supplied anchors @@ -134,13 +164,13 @@ bundle = sp_signals.build_dataset_with_custom_anchors( | `attribute_groups` | Attribute groups that provide the feature columns. Each attribute in these groups becomes a column in the final dataset. | `list[AttributeGroup]` | ✅ | | `anchors_table` | Table containing your pre-built anchor events | `WarehouseTable` | ✅ | | `anchors_have_label` | Whether the source table contains a `label` column. Set to `False` if your anchors are unlabeled. | `bool` | Default: `True` | -| `attributes_table` | Override the output table configuration for intermediate per-attribute-key tables (database, schema, table prefix) | `AttributesWarehouseTable` | Default: `None` | +| `attributes_table` | Override the output table location for the intermediate attribute tables. During execution, one table is created per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values. | `AttributesWarehouseTable` | Default: `None` | | `dataset_table` | Override the output table location for the final assembled dataset | `WarehouseTable` | Default: `None` | | `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. | `int` | Default: derived from attribute periods | ## Inspect and save the SQL bundle -Both `build_dataset_with_session_anchors()` and `build_dataset_with_custom_anchors()` return a `DatasetBundle` containing the generated SQL files. You can inspect them, save them to disk, or execute them directly. +Both `build_dataset_with_session_anchors()` and `build_dataset_with_custom_anchors()` return a `DatasetBundle` containing the generated SQL files. Inspecting the generated SQL is useful for understanding exactly what queries will run against your warehouse, verifying the logic before execution, or sharing with your data team for review. You can save them to disk or execute them directly. ### Save SQL to disk @@ -158,11 +188,11 @@ This creates: ## Execute against your warehouse -Once you have a `DatasetBundle`, execute the SQL against your warehouse to produce the training dataset. The warehouse connection is separate from your Signals API credentials because the SQL runs directly against your data warehouse, not through the Signals API. +Once you have a `DatasetBundle`, execute the SQL against your warehouse to produce the training dataset. The dataset builder executes SQL directly against the same data warehouse that your Signals deployment reads from. The warehouse connection is configured separately from your Signals API credentials because the queries run against your warehouse, not through the Signals API. ### Snowflake -Snowflake supports key-pair authentication for programmatic access. Wrap your Snowflake connection in a `SnowflakeConnection` and pass it to `execute()`. +Snowflake supports [key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth) for programmatic access, where `private_key` is the DER-encoded private key from your Snowflake key pair. Wrap your Snowflake connection in a `SnowflakeConnection` and pass it to `execute()`. ```python import snowflake.connector @@ -180,7 +210,9 @@ sf_conn = SnowflakeConnection( result = bundle.execute(sf_conn) ``` -The `execute()` method runs three stages in order: +The `execute()` method returns an `ExecutionResult` that holds references to the tables created in your warehouse. You can then call `to_pandas()` on the result to fetch the final dataset as a DataFrame. + +It runs three stages in order: 1. Creates the anchors table (`signals_anchors`) 2. Creates one attribute table per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values From 40e9b4757ffa045e62bd58fb7442532e5eb1556e Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Mon, 27 Jul 2026 11:07:18 +0100 Subject: [PATCH 06/10] add example notebook --- docs/signals/ml-training-datasets/index.md | 2 + .../notebooks/signals-dataset-builder.ipynb | 355 ++++++++++++++++++ 2 files changed, 357 insertions(+) create mode 100644 static/notebooks/signals-dataset-builder.ipynb diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md index 2008272db..ece25c350 100644 --- a/docs/signals/ml-training-datasets/index.md +++ b/docs/signals/ml-training-datasets/index.md @@ -11,6 +11,8 @@ The Signals dataset builder lets you create labeled training datasets for machin Training a machine learning model on behavioral data requires a dataset where each row represents a moment in time, with features computed only from events that had occurred up to that point. Building these datasets manually is error-prone: it is easy to accidentally include future information (data leakage), which produces models that perform well in testing but fail in production. The dataset builder automates this process, enforcing point-in-time correctness by construction. +To follow along with a working example, open the [dataset builder notebook](https://colab.research.google.com/github/snowplow/documentation/blob/signals-dataset-builder/static/notebooks/signals-dataset-builder.ipynb) in Google Colab. + Start by [connecting to Signals](/docs/signals/connection/index.md) to create a `Signals` client object. ## Why point-in-time correctness matters diff --git a/static/notebooks/signals-dataset-builder.ipynb b/static/notebooks/signals-dataset-builder.ipynb new file mode 100644 index 000000000..86343949e --- /dev/null +++ b/static/notebooks/signals-dataset-builder.ipynb @@ -0,0 +1,355 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "656e9de5", + "metadata": {}, + "source": [ + "# Signals dataset builder — ML training dataset example\n", + "\n", + "This notebook walks through the Signals **dataset builder**: turning your Snowplow event data into a **labeled, point-in-time-correct** training dataset for a machine learning model.\n", + "\n", + "**Scenario:** predict whether an ecommerce session will convert (a `transaction`) from behaviour seen *so far* in the session. Features are computed from the same attribute-group definitions that power real-time Signals, so training and serving use identical feature logic.\n", + "\n", + "**Why point-in-time correctness matters:** at serving time the model only sees events that have happened so far in the session. The builder computes each feature using only events *before* the anchor timestamp, which prevents data leakage (the most common cause of models that test well but fail in production).\n", + "\n", + "**Workflow:**\n", + "1. Define attribute groups with the features to learn from\n", + "2. Build the dataset SQL bundle (session anchors or custom anchors)\n", + "3. Inspect / save the generated SQL\n", + "4. Execute against your warehouse\n", + "\n", + "Docs: https://docs.snowplow.io/docs/signals/ml-training-datasets/" + ] + }, + { + "cell_type": "markdown", + "id": "b0933ff8", + "metadata": {}, + "source": [ + "## 1. Install the SDK" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "063a017f", + "metadata": {}, + "outputs": [], + "source": [ + "%pip install snowplow-signals==0.4.6rc1 python-dotenv snowflake-connector-python cryptography pandas" + ] + }, + { + "cell_type": "markdown", + "id": "ec80246b", + "metadata": {}, + "source": [ + "## 2. Connect to Signals\n", + "\n", + "Loads the four Signals credentials from `.env` and creates the `Signals` client. The client is used to build the dataset SQL bundle from your attribute-group definitions.\n", + "\n", + "See `.env.example` for the required variables." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "62cec6fd", + "metadata": {}, + "outputs": [], + "source": [ + "import os\n", + "from dotenv import load_dotenv\n", + "from snowplow_signals import Signals\n", + "\n", + "load_dotenv()\n", + "\n", + "sp_signals = Signals(\n", + " api_url=os.environ[\"SP_API_URL\"],\n", + " api_key=os.environ[\"SP_API_KEY\"],\n", + " api_key_id=os.environ[\"SP_API_KEY_ID\"],\n", + " org_id=os.environ[\"SP_ORG_ID\"],\n", + ")\n", + "sp_signals" + ] + }, + { + "cell_type": "markdown", + "id": "ff99684e", + "metadata": {}, + "source": [ + "## 3. Define the features\n", + "\n", + "Each attribute becomes a column in the training dataset. Here we count product views and add-to-cart actions within the session, filtered from the ecommerce action event by its `type` property.\n", + "\n", + "You don't need to publish these to build a dataset — the builder generates SQL directly from the definitions. Publishing is what you'd do later to serve the same features in real time." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "8cdb5e83", + "metadata": {}, + "outputs": [], + "source": [ + "from datetime import timedelta\n", + "from snowplow_signals import Attribute, Event, EventProperty, Criteria, Criterion\n", + "\n", + "# The Snowplow ecommerce action event carries an action `type` (product_view, add_to_cart, transaction, ...)\n", + "ecommerce_action = Event(\n", + " name=\"snowplow_ecommerce_action\",\n", + " vendor=\"com.snowplowanalytics.snowplow.ecommerce\",\n", + " version=\"1-0-2\",\n", + ")\n", + "\n", + "\n", + "def action_type_is(value: str) -> Criteria:\n", + " \"\"\"Filter ecommerce action events to a single action type.\"\"\"\n", + " return Criteria(\n", + " all=[\n", + " Criterion.eq(\n", + " EventProperty(\n", + " vendor=\"com.snowplowanalytics.snowplow.ecommerce\",\n", + " name=\"snowplow_ecommerce_action\",\n", + " major_version=1,\n", + " path=\"type\",\n", + " ),\n", + " value,\n", + " )\n", + " ]\n", + " )\n", + "\n", + "\n", + "product_view_count = Attribute(\n", + " name=\"product_view_count\",\n", + " description=\"Product views in the session so far\",\n", + " events=[ecommerce_action],\n", + " aggregation=\"counter\",\n", + " type=\"int32\",\n", + " period=timedelta(hours=1),\n", + " criteria=action_type_is(\"product_view\"),\n", + ")\n", + "\n", + "add_to_cart_count = Attribute(\n", + " name=\"add_to_cart_count\",\n", + " description=\"Add-to-cart actions in the session so far\",\n", + " events=[ecommerce_action],\n", + " aggregation=\"counter\",\n", + " type=\"int32\",\n", + " period=timedelta(hours=1),\n", + " criteria=action_type_is(\"add_to_cart\"),\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "855643d4", + "metadata": {}, + "source": [ + "### Group the features\n", + "\n", + "Attributes live inside an attribute group, keyed by `domain_sessionid` so features are computed per session." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5405b93b", + "metadata": {}, + "outputs": [], + "source": [ + "from snowplow_signals import StreamAttributeGroup, domain_sessionid\n", + "\n", + "ecommerce_group = StreamAttributeGroup(\n", + " name=\"ecommerce_session_features\",\n", + " version=1,\n", + " attribute_key=domain_sessionid,\n", + " owner=\"jack.keene@snowplowanalytics.com\",\n", + " attributes=[product_view_count, add_to_cart_count],\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "4180b0e9", + "metadata": {}, + "source": [ + "## 4. Build the dataset with session anchors\n", + "\n", + "`build_dataset_with_session_anchors()` scans every session in the training window and labels it: sessions where the **goal** (a `transaction`) occurred get a positive anchor (label=1) at the goal event; sessions without it get negative anchors (label=0) at random events, then downsampled to `max_negative_ratio`.\n", + "\n", + "Key knobs used below:\n", + "- `min_events=3` — skip anchors with too little prior signal\n", + "- `max_anchors_per_session=1` — one training row per session, so long sessions don't dominate\n", + "- `max_negative_ratio=1.0` — a balanced 1:1 positive/negative dataset\n", + "\n", + "This returns a `DatasetBundle` of generated SQL — nothing runs against your warehouse yet." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9351713a", + "metadata": {}, + "outputs": [], + "source": [ + "from datetime import datetime, timezone\n", + "from snowplow_signals import TrainingSpan\n", + "\n", + "bundle = sp_signals.build_dataset_with_session_anchors(\n", + " attribute_groups=[ecommerce_group],\n", + " goal_criteria=action_type_is(\"transaction\"),\n", + " training_span=TrainingSpan(\n", + " start_time=datetime(2024, 1, 1, tzinfo=timezone.utc),\n", + " end_time=datetime(2024, 4, 1, tzinfo=timezone.utc),\n", + " ),\n", + " min_events=3,\n", + " max_anchors_per_session=1,\n", + " max_negative_ratio=1.0,\n", + ")\n", + "bundle" + ] + }, + { + "cell_type": "markdown", + "id": "0959fc83", + "metadata": {}, + "source": [ + "## 5. Inspect / save the SQL bundle\n", + "\n", + "Write the generated SQL, a `manifest.json`, and a `README.md` to disk so you can review the exact queries or commit them for your data team before running anything." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ba6d06e3", + "metadata": {}, + "outputs": [], + "source": [ + "bundle.save_to(\"./dataset_output\")\n", + "print(\"SQL bundle written to ./dataset_output\")" + ] + }, + { + "cell_type": "markdown", + "id": "7c238640", + "metadata": {}, + "source": [ + "## 6. Execute against your warehouse (Snowflake)\n", + "\n", + "The builder runs the SQL directly against the same warehouse your Signals deployment reads from. The warehouse connection is **separate** from your Signals API credentials.\n", + "\n", + "Snowflake uses [key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth): `private_key` must be the DER-encoded private key. The helper below loads a PEM key file (path/passphrase from env) and converts it to DER." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "98a05f24", + "metadata": {}, + "outputs": [], + "source": [ + "from cryptography.hazmat.primitives import serialization\n", + "\n", + "\n", + "def load_private_key_der() -> bytes:\n", + " passphrase = os.environ.get(\"SNOWFLAKE_PRIVATE_KEY_PASSPHRASE\")\n", + " with open(os.environ[\"SNOWFLAKE_PRIVATE_KEY_PATH\"], \"rb\") as f:\n", + " private_key = serialization.load_pem_private_key(\n", + " f.read(),\n", + " password=passphrase.encode() if passphrase else None,\n", + " )\n", + " return private_key.private_bytes(\n", + " encoding=serialization.Encoding.DER,\n", + " format=serialization.PrivateFormat.PKCS8,\n", + " encryption_algorithm=serialization.NoEncryption(),\n", + " )\n", + "\n", + "\n", + "private_key_der = load_private_key_der()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c07aaddc", + "metadata": {}, + "outputs": [], + "source": [ + "import snowflake.connector\n", + "from snowplow_signals.execution.snowflake import SnowflakeConnection\n", + "\n", + "sf_conn = SnowflakeConnection(\n", + " snowflake.connector.connect(\n", + " account=os.environ[\"SNOWFLAKE_ACCOUNT\"],\n", + " user=os.environ[\"SNOWFLAKE_USER\"],\n", + " warehouse=os.environ[\"SNOWFLAKE_WAREHOUSE\"],\n", + " private_key=private_key_der,\n", + " )\n", + ")\n", + "\n", + "# Runs three stages: anchors -> per-key attribute tables -> assembled training dataset\n", + "result = bundle.execute(sf_conn)" + ] + }, + { + "cell_type": "markdown", + "id": "d01c84b9", + "metadata": {}, + "source": [ + "## 7. Get the training dataset as a DataFrame\n", + "\n", + "`to_pandas()` pulls the assembled dataset (one row per anchor; columns for the attribute key, anchor timestamp, label, and every attribute). Fetches up to 10,000 rows by default." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "35923045", + "metadata": {}, + "outputs": [], + "source": [ + "df = result.to_pandas()\n", + "print(df.shape)\n", + "df.head()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Alternative: user-supplied anchors\n", + "\n", + "If you already have a labeled anchor table (columns: your attribute key e.g. `domain_sessionid`, `anchor_ts`, and optionally `label`), use `build_dataset_with_custom_anchors()` instead of the session-anchor method. Everything downstream (`save_to`, `execute`, `to_pandas`) is identical.\n", + "\n", + "```python\n", + "from snowplow_signals import WarehouseTable\n", + "\n", + "bundle = sp_signals.build_dataset_with_custom_anchors(\n", + " attribute_groups=[ecommerce_group],\n", + " anchors_table=WarehouseTable(\n", + " database=\"analytics\",\n", + " schema=\"ml\",\n", + " table=\"my_anchor_events\",\n", + " ),\n", + " anchors_have_label=True,\n", + ")\n", + "```" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} From 1df6b11e61461b13c7ad7c3b3ecb962cc697747d Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Mon, 27 Jul 2026 11:32:13 +0100 Subject: [PATCH 07/10] use google colab --- .../notebooks/signals-dataset-builder.ipynb | 76 ++----------------- 1 file changed, 7 insertions(+), 69 deletions(-) diff --git a/static/notebooks/signals-dataset-builder.ipynb b/static/notebooks/signals-dataset-builder.ipynb index 86343949e..165b15d82 100644 --- a/static/notebooks/signals-dataset-builder.ipynb +++ b/static/notebooks/signals-dataset-builder.ipynb @@ -36,21 +36,13 @@ "id": "063a017f", "metadata": {}, "outputs": [], - "source": [ - "%pip install snowplow-signals==0.4.6rc1 python-dotenv snowflake-connector-python cryptography pandas" - ] + "source": "%pip install snowplow-signals==0.4.6rc1 snowflake-connector-python cryptography pandas" }, { "cell_type": "markdown", "id": "ec80246b", "metadata": {}, - "source": [ - "## 2. Connect to Signals\n", - "\n", - "Loads the four Signals credentials from `.env` and creates the `Signals` client. The client is used to build the dataset SQL bundle from your attribute-group definitions.\n", - "\n", - "See `.env.example` for the required variables." - ] + "source": "## 2. Connect to Signals\n\nLoads the four Signals credentials from [Google Colab secrets](https://x.com/GoogleColab/status/1719798406195867814) and creates the `Signals` client. The client is used to build the dataset SQL bundle from your attribute-group definitions.\n\nAdd the following secrets in the Colab sidebar (key icon): `SP_API_URL`, `SP_API_KEY`, `SP_API_KEY_ID`, `SP_ORG_ID`." }, { "cell_type": "code", @@ -58,21 +50,7 @@ "id": "62cec6fd", "metadata": {}, "outputs": [], - "source": [ - "import os\n", - "from dotenv import load_dotenv\n", - "from snowplow_signals import Signals\n", - "\n", - "load_dotenv()\n", - "\n", - "sp_signals = Signals(\n", - " api_url=os.environ[\"SP_API_URL\"],\n", - " api_key=os.environ[\"SP_API_KEY\"],\n", - " api_key_id=os.environ[\"SP_API_KEY_ID\"],\n", - " org_id=os.environ[\"SP_ORG_ID\"],\n", - ")\n", - "sp_signals" - ] + "source": "from google.colab import userdata\nfrom snowplow_signals import Signals\n\nsp_signals = Signals(\n api_url=userdata.get(\"SP_API_URL\"),\n api_key=userdata.get(\"SP_API_KEY\"),\n api_key_id=userdata.get(\"SP_API_KEY_ID\"),\n org_id=userdata.get(\"SP_ORG_ID\"),\n)\nsp_signals" }, { "cell_type": "markdown", @@ -236,13 +214,7 @@ "cell_type": "markdown", "id": "7c238640", "metadata": {}, - "source": [ - "## 6. Execute against your warehouse (Snowflake)\n", - "\n", - "The builder runs the SQL directly against the same warehouse your Signals deployment reads from. The warehouse connection is **separate** from your Signals API credentials.\n", - "\n", - "Snowflake uses [key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth): `private_key` must be the DER-encoded private key. The helper below loads a PEM key file (path/passphrase from env) and converts it to DER." - ] + "source": "## 6. Execute against your warehouse (Snowflake)\n\nThe builder runs the SQL directly against the same warehouse your Signals deployment reads from. The warehouse connection is **separate** from your Signals API credentials.\n\nSnowflake uses [key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth): `private_key` must be the DER-encoded private key. The helper below loads a PEM key from a Colab secret and converts it to DER.\n\nAdd the following secrets in the Colab sidebar: `SNOWFLAKE_PRIVATE_KEY_PEM`, `SNOWFLAKE_PRIVATE_KEY_PASSPHRASE` (optional), `SNOWFLAKE_ACCOUNT`, `SNOWFLAKE_USER`, `SNOWFLAKE_WAREHOUSE`." }, { "cell_type": "code", @@ -250,26 +222,7 @@ "id": "98a05f24", "metadata": {}, "outputs": [], - "source": [ - "from cryptography.hazmat.primitives import serialization\n", - "\n", - "\n", - "def load_private_key_der() -> bytes:\n", - " passphrase = os.environ.get(\"SNOWFLAKE_PRIVATE_KEY_PASSPHRASE\")\n", - " with open(os.environ[\"SNOWFLAKE_PRIVATE_KEY_PATH\"], \"rb\") as f:\n", - " private_key = serialization.load_pem_private_key(\n", - " f.read(),\n", - " password=passphrase.encode() if passphrase else None,\n", - " )\n", - " return private_key.private_bytes(\n", - " encoding=serialization.Encoding.DER,\n", - " format=serialization.PrivateFormat.PKCS8,\n", - " encryption_algorithm=serialization.NoEncryption(),\n", - " )\n", - "\n", - "\n", - "private_key_der = load_private_key_der()" - ] + "source": "from cryptography.hazmat.primitives import serialization\n\n\ndef load_private_key_der() -> bytes:\n passphrase = userdata.get(\"SNOWFLAKE_PRIVATE_KEY_PASSPHRASE\")\n pem_data = userdata.get(\"SNOWFLAKE_PRIVATE_KEY_PEM\").encode()\n private_key = serialization.load_pem_private_key(\n pem_data,\n password=passphrase.encode() if passphrase else None,\n )\n return private_key.private_bytes(\n encoding=serialization.Encoding.DER,\n format=serialization.PrivateFormat.PKCS8,\n encryption_algorithm=serialization.NoEncryption(),\n )\n\n\nprivate_key_der = load_private_key_der()" }, { "cell_type": "code", @@ -277,22 +230,7 @@ "id": "c07aaddc", "metadata": {}, "outputs": [], - "source": [ - "import snowflake.connector\n", - "from snowplow_signals.execution.snowflake import SnowflakeConnection\n", - "\n", - "sf_conn = SnowflakeConnection(\n", - " snowflake.connector.connect(\n", - " account=os.environ[\"SNOWFLAKE_ACCOUNT\"],\n", - " user=os.environ[\"SNOWFLAKE_USER\"],\n", - " warehouse=os.environ[\"SNOWFLAKE_WAREHOUSE\"],\n", - " private_key=private_key_der,\n", - " )\n", - ")\n", - "\n", - "# Runs three stages: anchors -> per-key attribute tables -> assembled training dataset\n", - "result = bundle.execute(sf_conn)" - ] + "source": "import snowflake.connector\nfrom snowplow_signals.execution.snowflake import SnowflakeConnection\n\nsf_conn = SnowflakeConnection(\n snowflake.connector.connect(\n account=userdata.get(\"SNOWFLAKE_ACCOUNT\"),\n user=userdata.get(\"SNOWFLAKE_USER\"),\n warehouse=userdata.get(\"SNOWFLAKE_WAREHOUSE\"),\n private_key=private_key_der,\n )\n)\n\n# Runs three stages: anchors -> per-key attribute tables -> assembled training dataset\nresult = bundle.execute(sf_conn)" }, { "cell_type": "markdown", @@ -352,4 +290,4 @@ }, "nbformat": 4, "nbformat_minor": 5 -} +} \ No newline at end of file From 5b99ac921321c3c95e470f5dc39582bbf05194bf Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Mon, 27 Jul 2026 17:11:52 +0100 Subject: [PATCH 08/10] pr comments --- docs/signals/ml-training-datasets/index.md | 39 ++++++++++++++++------ 1 file changed, 28 insertions(+), 11 deletions(-) diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md index ece25c350..99e865499 100644 --- a/docs/signals/ml-training-datasets/index.md +++ b/docs/signals/ml-training-datasets/index.md @@ -40,21 +40,21 @@ The dataset builder produces a training dataset in three stages: flowchart LR subgraph stage1["1. Anchors"] direction TB - A1["Scan sessions in\ntraining window"] --> A2{"Goal event\noccurred?"} - A2 -- Yes --> A3["Positive anchor\n(label = 1)"] - A2 -- No --> A4["Negative anchor\n(label = 0)"] - A3 --> A5["Downsample\nnegatives"] + A1["Scan sessions in
training window"] --> A2{"Goal event
occurred?"} + A2 -- Yes --> A3["Positive anchor
(label = 1)"] + A2 -- No --> A4["Negative anchor
(label = 0)"] + A3 --> A5["Downsample
negatives"] A4 --> A5 end subgraph stage2["2. Attributes"] direction TB - B1["For each anchor,\nfind preceding events"] --> B2["Compute attribute\nvalues using only\nevents before\nanchor timestamp"] + B1["For each anchor,
find preceding events"] --> B2["Compute attribute
values using only
events before
anchor timestamp"] end subgraph stage3["3. Assembly"] direction TB - C1["Join anchors +\nattributes into\nlabeled dataset"] --> C2["One row per anchor\nOne column per attribute"] + C1["Join anchors +
attributes into
labeled dataset"] --> C2["One row per anchor
One column per attribute"] end stage1 --> stage2 --> stage3 @@ -125,11 +125,28 @@ bundle = sp_signals.build_dataset_with_session_anchors( | `max_anchors_per_session` | Maximum anchor events per session. `None` for unlimited. Set this to limit overrepresentation of long sessions, for example set to `1` to ensure each session contributes at most one training example. | `int` or `None` | Default: `None` | | `max_negative_ratio` | Maximum ratio of negative to positive anchors. Negative anchors are downsampled to this ratio. Lower values produce more balanced datasets; higher values preserve more data. For example, set to `1.0` for a balanced 1:1 dataset. | `float` | Default: `5.0` | | `excluded_events` | Events to exclude from anchor generation. By default, `page_ping` events are excluded because they do not represent meaningful user actions. | `list` | Default: `page_ping` events excluded | -| `anchors_table` | Custom output table location for the anchors table. By default, a table named `signals_anchors` is created in your warehouse. | `WarehouseTable` | Default: `None` | -| `attributes_table` | Override the output table location for the intermediate attribute tables. During execution, one table is created per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values. | `AttributesWarehouseTable` | Default: `None` | -| `dataset_table` | Custom output table location for the final assembled dataset. By default, a table named `signals_training_dataset` is created in your warehouse. | `WarehouseTable` | Default: `None` | +| `anchors_table` | Override the output location for the anchors table. Only the `table` field is required - `database` and `schema` default to the output database and schema from your [Signals warehouse connection](/docs/signals/setup/index.md). The default table name is `signals_anchors`. | `WarehouseTable` | Default: `None` | +| `attributes_table` | Override the output location for the intermediate attribute tables. During execution, one table is created per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values. Supports `database`, `schema`, and `table_prefix` fields - all default to the output database and schema from your [Signals warehouse connection](/docs/signals/setup/index.md). | `AttributesWarehouseTable` | Default: `None` | +| `dataset_table` | Override the output location for the final assembled dataset. Only the `table` field is required - `database` and `schema` default to the output database and schema from your [Signals warehouse connection](/docs/signals/setup/index.md). The default table name is `signals_training_dataset`. | `WarehouseTable` | Default: `None` | | `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. Set a lower value to narrow the event window, or a higher value to include older events. | `int` | Default: derived from attribute periods | +#### Customizing output tables + +By default, the dataset builder creates tables using the output database and schema from your [Signals warehouse connection](/docs/signals/setup/index.md). To write tables to a different location, pass a `WarehouseTable`. Only `table` is required - `database` and `schema` are optional overrides. + +```python +from snowplow_signals import WarehouseTable + +bundle = sp_signals.build_dataset_with_session_anchors( + ... + dataset_table=WarehouseTable( + table="my_training_dataset", # required + database="analytics", # optional, defaults to Signals connection + schema="ml", # optional, defaults to Signals connection + ), +) +``` + ### User-supplied anchors If you already have a table of labeled anchor events, use `build_dataset_with_custom_anchors()` instead. Your table must contain the following columns: @@ -166,8 +183,8 @@ bundle = sp_signals.build_dataset_with_custom_anchors( | `attribute_groups` | Attribute groups that provide the feature columns. Each attribute in these groups becomes a column in the final dataset. | `list[AttributeGroup]` | ✅ | | `anchors_table` | Table containing your pre-built anchor events | `WarehouseTable` | ✅ | | `anchors_have_label` | Whether the source table contains a `label` column. Set to `False` if your anchors are unlabeled. | `bool` | Default: `True` | -| `attributes_table` | Override the output table location for the intermediate attribute tables. During execution, one table is created per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values. | `AttributesWarehouseTable` | Default: `None` | -| `dataset_table` | Override the output table location for the final assembled dataset | `WarehouseTable` | Default: `None` | +| `attributes_table` | Override the output location for the intermediate attribute tables. During execution, one table is created per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values. Supports `database`, `schema`, and `table_prefix` fields - all default to the output database and schema from your [Signals warehouse connection](/docs/signals/setup/index.md). | `AttributesWarehouseTable` | Default: `None` | +| `dataset_table` | Override the output location for the final assembled dataset. Only the `table` field is required - `database` and `schema` default to the output database and schema from your [Signals warehouse connection](/docs/signals/setup/index.md). | `WarehouseTable` | Default: `None` | | `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. | `int` | Default: derived from attribute periods | ## Inspect and save the SQL bundle From e504aec7dba68833334ef40a61ea45537ddd427a Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Mon, 27 Jul 2026 17:54:00 +0100 Subject: [PATCH 09/10] add more complete criterion example --- docs/signals/ml-training-datasets/index.md | 27 ++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md index 99e865499..5f0f70f11 100644 --- a/docs/signals/ml-training-datasets/index.md +++ b/docs/signals/ml-training-datasets/index.md @@ -84,20 +84,43 @@ Signals provides two approaches: session anchors (automatically derived from you Use `build_dataset_with_session_anchors()` to automatically generate anchors from your event data. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window and labels each based on whether the goal was achieved. For positive sessions, anchors are placed at the goal event itself. For negative sessions (where the goal was never achieved), anchors are placed at randomly selected events within the session. Negative anchors are then downsampled according to `max_negative_ratio` to avoid class imbalance. +The `goal_criteria` argument uses the same `Criteria` and `Criterion` classes as [attribute criteria](/docs/signals/attributes/attributes/index.md). You can filter on atomic fields, event properties, or entity properties. + +To match on an event name (an atomic field): + ```python from datetime import datetime, timezone from snowplow_signals import ( + AtomicProperty, Criteria, Criterion, - EventProperty, TrainingSpan, ) bundle = sp_signals.build_dataset_with_session_anchors( attribute_groups=[my_attribute_group], goal_criteria=Criteria( - any=[ + all=[ + Criterion.eq(AtomicProperty(name="event"), "transaction"), + ] + ), + training_span=TrainingSpan( + start_time=datetime(2024, 1, 1, tzinfo=timezone.utc), + end_time=datetime(2024, 4, 1, tzinfo=timezone.utc), + ), +) +``` + +To filter on a property within a self-describing event: + +```python +from snowplow_signals import Criteria, Criterion, EventProperty, TrainingSpan + +bundle = sp_signals.build_dataset_with_session_anchors( + attribute_groups=[my_attribute_group], + goal_criteria=Criteria( + all=[ Criterion.eq( EventProperty( vendor="com.snowplowanalytics.snowplow.ecommerce", From 7b76e4c195776226935891005bea9f1328795d17 Mon Sep 17 00:00:00 2001 From: Jack Keene Date: Thu, 30 Jul 2026 16:44:18 +0100 Subject: [PATCH 10/10] add SS execution docs --- docs/signals/ml-training-datasets/index.md | 118 ++++++++-------- .../notebooks/signals-dataset-builder.ipynb | 126 ++---------------- 2 files changed, 73 insertions(+), 171 deletions(-) diff --git a/docs/signals/ml-training-datasets/index.md b/docs/signals/ml-training-datasets/index.md index 5f0f70f11..96ba081a3 100644 --- a/docs/signals/ml-training-datasets/index.md +++ b/docs/signals/ml-training-datasets/index.md @@ -2,8 +2,8 @@ title: "Create ML training datasets" sidebar_position: 32 sidebar_label: "ML training datasets" -description: "Build labeled, point-in-time correct training datasets for machine learning models using the Signals Python SDK. Define anchor events, generate SQL, and execute against your warehouse." -keywords: ["training datasets", "machine learning", "dataset builder", "anchors", "signals python sdk", "snowflake", "ML"] +description: "Build labeled, point-in-time correct training datasets for machine learning models using the Signals Python SDK. Define anchor events, build datasets, and retrieve results." +keywords: ["training datasets", "machine learning", "dataset builder", "anchors", "signals python sdk", "ML"] date: "2026-07-15" --- @@ -26,11 +26,10 @@ The dataset builder prevents this by computing each attribute value using only e Building and using an ML training dataset follows this process: 1. [Define your attribute groups](/docs/signals/attributes/attribute-groups/index.md) with the features you want your model to learn from (e.g. `product_view_count`, `add_to_cart_count`) -2. Build the dataset SQL bundle by calling the appropriate method - `build_dataset_with_session_anchors()` for auto-generated anchors, or `build_dataset_with_custom_anchors()` for a pre-existing anchor table -3. Optionally inspect or save the generated SQL -4. Execute the SQL against your warehouse to produce the training dataset -5. Train your model on the resulting DataFrame -6. Deploy your model, serving it the same attributes in real time via [Retrieve attributes](/docs/signals/applications/retrieve-attributes/index.md) +2. Submit a dataset run by calling the appropriate method - `submit_dataset_run_with_session_anchors()` for auto-generated anchors, or `submit_dataset_run_with_custom_anchors()` for a pre-existing anchor table +3. Poll for completion with `get_dataset_run_status()`, then retrieve a preview with `get_dataset_run_preview()` +4. Train your model on the resulting DataFrame +5. Deploy your model, serving it the same attributes in real time via [Retrieve attributes](/docs/signals/applications/retrieve-attributes/index.md) ## How it works @@ -82,7 +81,7 @@ Signals provides two approaches: session anchors (automatically derived from you ### Session anchors -Use `build_dataset_with_session_anchors()` to automatically generate anchors from your event data. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window and labels each based on whether the goal was achieved. For positive sessions, anchors are placed at the goal event itself. For negative sessions (where the goal was never achieved), anchors are placed at randomly selected events within the session. Negative anchors are then downsampled according to `max_negative_ratio` to avoid class imbalance. +Use `submit_dataset_run_with_session_anchors()` to automatically generate anchors from your event data and build the training dataset server-side. You specify a goal (the criteria that define a positive outcome) and a time window to scan. Signals scans all sessions in the training window and labels each based on whether the goal was achieved. For positive sessions, anchors are placed at the goal event itself. For negative sessions (where the goal was never achieved), anchors are placed at randomly selected events within the session. Negative anchors are then downsampled according to `max_negative_ratio` to avoid class imbalance. The `goal_criteria` argument uses the same `Criteria` and `Criterion` classes as [attribute criteria](/docs/signals/attributes/attributes/index.md). You can filter on atomic fields, event properties, or entity properties. @@ -98,7 +97,7 @@ from snowplow_signals import ( TrainingSpan, ) -bundle = sp_signals.build_dataset_with_session_anchors( +run = sp_signals.submit_dataset_run_with_session_anchors( attribute_groups=[my_attribute_group], goal_criteria=Criteria( all=[ @@ -117,7 +116,7 @@ To filter on a property within a self-describing event: ```python from snowplow_signals import Criteria, Criterion, EventProperty, TrainingSpan -bundle = sp_signals.build_dataset_with_session_anchors( +run = sp_signals.submit_dataset_run_with_session_anchors( attribute_groups=[my_attribute_group], goal_criteria=Criteria( all=[ @@ -160,7 +159,7 @@ By default, the dataset builder creates tables using the output database and sch ```python from snowplow_signals import WarehouseTable -bundle = sp_signals.build_dataset_with_session_anchors( +run = sp_signals.submit_dataset_run_with_session_anchors( ... dataset_table=WarehouseTable( table="my_training_dataset", # required @@ -172,7 +171,7 @@ bundle = sp_signals.build_dataset_with_session_anchors( ### User-supplied anchors -If you already have a table of labeled anchor events, use `build_dataset_with_custom_anchors()` instead. Your table must contain the following columns: +If you already have a table of labeled anchor events, use `submit_dataset_run_with_custom_anchors()` instead. Your table must contain the following columns: | Column | Type | Description | | --- | --- | --- | @@ -190,7 +189,7 @@ For example, if your attribute groups use `domain_sessionid` as the attribute ke ```python from snowplow_signals import WarehouseTable -bundle = sp_signals.build_dataset_with_custom_anchors( +run = sp_signals.submit_dataset_run_with_custom_anchors( attribute_groups=[my_attribute_group], anchors_table=WarehouseTable( database="analytics", @@ -210,64 +209,41 @@ bundle = sp_signals.build_dataset_with_custom_anchors( | `dataset_table` | Override the output location for the final assembled dataset. Only the `table` field is required - `database` and `schema` default to the output database and schema from your [Signals warehouse connection](/docs/signals/setup/index.md). | `WarehouseTable` | Default: `None` | | `max_lookback_days` | How far back from each anchor timestamp to look for events when computing attributes. By default, this is derived from the longest period defined across your attributes. | `int` | Default: derived from attribute periods | -## Inspect and save the SQL bundle +## Poll for completion and retrieve results -Both `build_dataset_with_session_anchors()` and `build_dataset_with_custom_anchors()` return a `DatasetBundle` containing the generated SQL files. Inspecting the generated SQL is useful for understanding exactly what queries will run against your warehouse, verifying the logic before execution, or sharing with your data team for review. You can save them to disk or execute them directly. +The `submit_dataset_run_with_session_anchors()` and `submit_dataset_run_with_custom_anchors()` methods return a `DatasetRunResponse` immediately. The dataset is built server-side, so you need to poll for completion before retrieving results. -### Save SQL to disk +### Check run status -Use `save_to()` to write the generated SQL files, a manifest, and a README to a directory. This is useful for reviewing the SQL before execution or committing it to version control. +Use `get_dataset_run_status()` to check whether the run has finished. The `status` field is one of `pending`, `success`, or `failed`. ```python -bundle.save_to("./dataset_output") +status = sp_signals.get_dataset_run_status(run.id) +print(status.status) # "pending", "success", or "failed" ``` -This creates: - -- Individual SQL files for each stage (`signals_anchors.sql`, `signals_attributes_domain_sessionid.sql`, `signals_training_dataset.sql`) -- `manifest.json` with input configuration and output table mappings -- `README.md` documenting the execution order - -## Execute against your warehouse - -Once you have a `DatasetBundle`, execute the SQL against your warehouse to produce the training dataset. The dataset builder executes SQL directly against the same data warehouse that your Signals deployment reads from. The warehouse connection is configured separately from your Signals API credentials because the queries run against your warehouse, not through the Signals API. - -### Snowflake - -Snowflake supports [key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth) for programmatic access, where `private_key` is the DER-encoded private key from your Snowflake key pair. Wrap your Snowflake connection in a `SnowflakeConnection` and pass it to `execute()`. +To wait for completion in a notebook or script: ```python -import snowflake.connector -from snowplow_signals.execution.snowflake import SnowflakeConnection - -sf_conn = SnowflakeConnection( - snowflake.connector.connect( - account=os.environ["SNOWFLAKE_ACCOUNT"], - user=os.environ["SNOWFLAKE_USER"], - warehouse=os.environ["SNOWFLAKE_WAREHOUSE"], - private_key=private_key_der, - ) -) - -result = bundle.execute(sf_conn) +import time +from snowplow_signals import DatasetRunStatus + +while True: + status = sp_signals.get_dataset_run_status(run.id) + if status.status == DatasetRunStatus.SUCCESS: + break + if status.status == DatasetRunStatus.FAILED: + raise RuntimeError("Dataset run failed") + time.sleep(5) ``` -The `execute()` method returns an `ExecutionResult` that holds references to the tables created in your warehouse. You can then call `to_pandas()` on the result to fetch the final dataset as a DataFrame. - -It runs three stages in order: - -1. Creates the anchors table (`signals_anchors`) -2. Creates one attribute table per attribute key (e.g. `signals_attributes_domain_sessionid`), joining each anchor with its point-in-time attribute values -3. Assembles the final training dataset (`signals_training_dataset`) by joining all tables together - -If a stage fails, it raises an `ExecutionError` with the name of the failed stage, the target table, and the underlying cause. - ### Get results as a DataFrame -After execution, call `to_pandas()` to pull the training dataset into a pandas DataFrame, ready for model training. By default, it fetches up to 10,000 rows. +Once the run status is `success`, call `get_dataset_run_preview()` to fetch a preview of the completed dataset. By default, it returns up to 100 rows. You can set `limit` up to 10,000. ```python -df = result.to_pandas() +preview = sp_signals.get_dataset_run_preview(run.id, limit=10000) +df = preview.to_pandas() ``` The resulting DataFrame contains one row per anchor, with columns for the attribute key, anchor timestamp, label, and every attribute from your attribute groups: @@ -277,3 +253,33 @@ The resulting DataFrame contains one row per anchor, with columns for the attrib | 0 | abc-123 | 2024-01-15 09:32:00 | 1 | 5 | 2 | | 1 | def-456 | 2024-01-15 10:01:00 | 0 | 3 | 0 | | 2 | ghi-789 | 2024-01-16 14:22:00 | 1 | 8 | 4 | + +The full dataset is also written to your warehouse at the table location shown in `run.dataset`. You can query this table directly for the complete result set. + +### Cancel a run + +To cancel a dataset build that is still in progress: + +```python +sp_signals.cancel_dataset_run(run.id) +``` + +## Inspect the generated SQL + +If you want to review the SQL that the dataset builder produces without executing it, use `build_dataset_with_session_anchors()` or `build_dataset_with_custom_anchors()`. These return a `DatasetBundle` containing the generated SQL files, which you can save to disk for inspection or manual execution. + +```python +bundle = sp_signals.build_dataset_with_session_anchors( + attribute_groups=[my_attribute_group], + goal_criteria=goal, + training_span=span, +) + +bundle.save_to("./dataset_output") +``` + +This creates: + +- Individual SQL files for each stage (`signals_anchors.sql`, `signals_attributes_domain_sessionid.sql`, `signals_training_dataset.sql`) +- `manifest.json` with input configuration and output table mappings +- `README.md` documenting the execution order diff --git a/static/notebooks/signals-dataset-builder.ipynb b/static/notebooks/signals-dataset-builder.ipynb index 165b15d82..40fc0611b 100644 --- a/static/notebooks/signals-dataset-builder.ipynb +++ b/static/notebooks/signals-dataset-builder.ipynb @@ -4,23 +4,7 @@ "cell_type": "markdown", "id": "656e9de5", "metadata": {}, - "source": [ - "# Signals dataset builder — ML training dataset example\n", - "\n", - "This notebook walks through the Signals **dataset builder**: turning your Snowplow event data into a **labeled, point-in-time-correct** training dataset for a machine learning model.\n", - "\n", - "**Scenario:** predict whether an ecommerce session will convert (a `transaction`) from behaviour seen *so far* in the session. Features are computed from the same attribute-group definitions that power real-time Signals, so training and serving use identical feature logic.\n", - "\n", - "**Why point-in-time correctness matters:** at serving time the model only sees events that have happened so far in the session. The builder computes each feature using only events *before* the anchor timestamp, which prevents data leakage (the most common cause of models that test well but fail in production).\n", - "\n", - "**Workflow:**\n", - "1. Define attribute groups with the features to learn from\n", - "2. Build the dataset SQL bundle (session anchors or custom anchors)\n", - "3. Inspect / save the generated SQL\n", - "4. Execute against your warehouse\n", - "\n", - "Docs: https://docs.snowplow.io/docs/signals/ml-training-datasets/" - ] + "source": "# Signals dataset builder - ML training dataset example\n\nThis notebook walks through the Signals **dataset builder**: turning your Snowplow event data into a **labeled, point-in-time-correct** training dataset for a machine learning model.\n\n**Scenario:** predict whether an ecommerce session will convert (a `transaction`) from behaviour seen *so far* in the session. Features are computed from the same attribute-group definitions that power real-time Signals, so training and serving use identical feature logic.\n\n**Why point-in-time correctness matters:** at serving time the model only sees events that have happened so far in the session. The builder computes each feature using only events *before* the anchor timestamp, which prevents data leakage (the most common cause of models that test well but fail in production).\n\n**Workflow:**\n1. Define attribute groups with the features to learn from\n2. Submit a dataset run (session anchors or custom anchors)\n3. Poll for completion and retrieve results\n4. Optionally inspect / save the generated SQL\n\nDocs: https://docs.snowplow.io/docs/signals/ml-training-datasets/" }, { "cell_type": "markdown", @@ -36,13 +20,13 @@ "id": "063a017f", "metadata": {}, "outputs": [], - "source": "%pip install snowplow-signals==0.4.6rc1 snowflake-connector-python cryptography pandas" + "source": "%pip install snowplow-signals==0.4.7 pandas" }, { "cell_type": "markdown", "id": "ec80246b", "metadata": {}, - "source": "## 2. Connect to Signals\n\nLoads the four Signals credentials from [Google Colab secrets](https://x.com/GoogleColab/status/1719798406195867814) and creates the `Signals` client. The client is used to build the dataset SQL bundle from your attribute-group definitions.\n\nAdd the following secrets in the Colab sidebar (key icon): `SP_API_URL`, `SP_API_KEY`, `SP_API_KEY_ID`, `SP_ORG_ID`." + "source": "## 2. Connect to Signals\n\nLoads the four Signals credentials from [Google Colab secrets](https://x.com/GoogleColab/status/1719798406195867814) and creates the `Signals` client. The client submits dataset runs and retrieves results via the Signals API.\n\nAdd the following secrets in the Colab sidebar (key icon): `SP_API_URL`, `SP_API_KEY`, `SP_API_KEY_ID`, `SP_ORG_ID`." }, { "cell_type": "code", @@ -56,13 +40,7 @@ "cell_type": "markdown", "id": "ff99684e", "metadata": {}, - "source": [ - "## 3. Define the features\n", - "\n", - "Each attribute becomes a column in the training dataset. Here we count product views and add-to-cart actions within the session, filtered from the ecommerce action event by its `type` property.\n", - "\n", - "You don't need to publish these to build a dataset — the builder generates SQL directly from the definitions. Publishing is what you'd do later to serve the same features in real time." - ] + "source": "## 3. Define the features\n\nEach attribute becomes a column in the training dataset. Here we count product views and add-to-cart actions within the session, filtered from the ecommerce action event by its `type` property.\n\nYou don't need to publish these to build a dataset - the builder works directly from the definitions. Publishing is what you'd do later to serve the same features in real time." }, { "cell_type": "code", @@ -152,18 +130,7 @@ "cell_type": "markdown", "id": "4180b0e9", "metadata": {}, - "source": [ - "## 4. Build the dataset with session anchors\n", - "\n", - "`build_dataset_with_session_anchors()` scans every session in the training window and labels it: sessions where the **goal** (a `transaction`) occurred get a positive anchor (label=1) at the goal event; sessions without it get negative anchors (label=0) at random events, then downsampled to `max_negative_ratio`.\n", - "\n", - "Key knobs used below:\n", - "- `min_events=3` — skip anchors with too little prior signal\n", - "- `max_anchors_per_session=1` — one training row per session, so long sessions don't dominate\n", - "- `max_negative_ratio=1.0` — a balanced 1:1 positive/negative dataset\n", - "\n", - "This returns a `DatasetBundle` of generated SQL — nothing runs against your warehouse yet." - ] + "source": "## 4. Submit a dataset run with session anchors\n\n`submit_dataset_run_with_session_anchors()` submits a dataset build for server-side execution. It scans every session in the training window and labels it: sessions where the **goal** (a `transaction`) occurred get a positive anchor (label=1) at the goal event; sessions without it get negative anchors (label=0) at random events, then downsampled to `max_negative_ratio`.\n\nKey knobs used below:\n- `min_events=3` - skip anchors with too little prior signal\n- `max_anchors_per_session=10` - up to 10 training rows per session, so long sessions don't dominate\n- `max_negative_ratio=1.0` - a balanced 1:1 positive/negative dataset\n\nThis returns a `DatasetRunResponse` immediately. The dataset is built asynchronously on the server." }, { "cell_type": "code", @@ -171,33 +138,13 @@ "id": "9351713a", "metadata": {}, "outputs": [], - "source": [ - "from datetime import datetime, timezone\n", - "from snowplow_signals import TrainingSpan\n", - "\n", - "bundle = sp_signals.build_dataset_with_session_anchors(\n", - " attribute_groups=[ecommerce_group],\n", - " goal_criteria=action_type_is(\"transaction\"),\n", - " training_span=TrainingSpan(\n", - " start_time=datetime(2024, 1, 1, tzinfo=timezone.utc),\n", - " end_time=datetime(2024, 4, 1, tzinfo=timezone.utc),\n", - " ),\n", - " min_events=3,\n", - " max_anchors_per_session=1,\n", - " max_negative_ratio=1.0,\n", - ")\n", - "bundle" - ] + "source": "from datetime import datetime, timezone\nfrom snowplow_signals import TrainingSpan\n\nrun = sp_signals.submit_dataset_run_with_session_anchors(\n attribute_groups=[ecommerce_group],\n goal_criteria=action_type_is(\"transaction\"),\n training_span=TrainingSpan(\n start_time=datetime(2024, 1, 1, tzinfo=timezone.utc),\n end_time=datetime(2024, 4, 1, tzinfo=timezone.utc),\n ),\n min_events=3,\n max_anchors_per_session=10,\n max_negative_ratio=1.0,\n)\nrun" }, { "cell_type": "markdown", "id": "0959fc83", "metadata": {}, - "source": [ - "## 5. Inspect / save the SQL bundle\n", - "\n", - "Write the generated SQL, a `manifest.json`, and a `README.md` to disk so you can review the exact queries or commit them for your data team before running anything." - ] + "source": "## 5. Poll for completion\n\nThe dataset is built server-side. Poll `get_dataset_run_status()` until the status is `success` or `failed`." }, { "cell_type": "code", @@ -205,42 +152,13 @@ "id": "ba6d06e3", "metadata": {}, "outputs": [], - "source": [ - "bundle.save_to(\"./dataset_output\")\n", - "print(\"SQL bundle written to ./dataset_output\")" - ] + "source": "import time\nfrom snowplow_signals import DatasetRunStatus\n\nwhile True:\n status = sp_signals.get_dataset_run_status(run.id)\n print(f\"Status: {status.status}\")\n if status.status == DatasetRunStatus.SUCCESS:\n break\n if status.status == DatasetRunStatus.FAILED:\n raise RuntimeError(\"Dataset run failed\")\n time.sleep(5)" }, { "cell_type": "markdown", "id": "7c238640", "metadata": {}, - "source": "## 6. Execute against your warehouse (Snowflake)\n\nThe builder runs the SQL directly against the same warehouse your Signals deployment reads from. The warehouse connection is **separate** from your Signals API credentials.\n\nSnowflake uses [key-pair authentication](https://docs.snowflake.com/en/user-guide/key-pair-auth): `private_key` must be the DER-encoded private key. The helper below loads a PEM key from a Colab secret and converts it to DER.\n\nAdd the following secrets in the Colab sidebar: `SNOWFLAKE_PRIVATE_KEY_PEM`, `SNOWFLAKE_PRIVATE_KEY_PASSPHRASE` (optional), `SNOWFLAKE_ACCOUNT`, `SNOWFLAKE_USER`, `SNOWFLAKE_WAREHOUSE`." - }, - { - "cell_type": "code", - "execution_count": null, - "id": "98a05f24", - "metadata": {}, - "outputs": [], - "source": "from cryptography.hazmat.primitives import serialization\n\n\ndef load_private_key_der() -> bytes:\n passphrase = userdata.get(\"SNOWFLAKE_PRIVATE_KEY_PASSPHRASE\")\n pem_data = userdata.get(\"SNOWFLAKE_PRIVATE_KEY_PEM\").encode()\n private_key = serialization.load_pem_private_key(\n pem_data,\n password=passphrase.encode() if passphrase else None,\n )\n return private_key.private_bytes(\n encoding=serialization.Encoding.DER,\n format=serialization.PrivateFormat.PKCS8,\n encryption_algorithm=serialization.NoEncryption(),\n )\n\n\nprivate_key_der = load_private_key_der()" - }, - { - "cell_type": "code", - "execution_count": null, - "id": "c07aaddc", - "metadata": {}, - "outputs": [], - "source": "import snowflake.connector\nfrom snowplow_signals.execution.snowflake import SnowflakeConnection\n\nsf_conn = SnowflakeConnection(\n snowflake.connector.connect(\n account=userdata.get(\"SNOWFLAKE_ACCOUNT\"),\n user=userdata.get(\"SNOWFLAKE_USER\"),\n warehouse=userdata.get(\"SNOWFLAKE_WAREHOUSE\"),\n private_key=private_key_der,\n )\n)\n\n# Runs three stages: anchors -> per-key attribute tables -> assembled training dataset\nresult = bundle.execute(sf_conn)" - }, - { - "cell_type": "markdown", - "id": "d01c84b9", - "metadata": {}, - "source": [ - "## 7. Get the training dataset as a DataFrame\n", - "\n", - "`to_pandas()` pulls the assembled dataset (one row per anchor; columns for the attribute key, anchor timestamp, label, and every attribute). Fetches up to 10,000 rows by default." - ] + "source": "## 6. Get the training dataset as a DataFrame\n\n`get_dataset_run_preview()` fetches a preview of the completed dataset. Call `to_pandas()` on the result to get a DataFrame (one row per anchor; columns for the attribute key, anchor timestamp, label, and every attribute). Set `limit` up to 10,000 rows." }, { "cell_type": "code", @@ -248,34 +166,12 @@ "id": "35923045", "metadata": {}, "outputs": [], - "source": [ - "df = result.to_pandas()\n", - "print(df.shape)\n", - "df.head()" - ] + "source": "preview = sp_signals.get_dataset_run_preview(run.id, limit=10000)\ndf = preview.to_pandas()\nprint(df.shape)\ndf.head()" }, { "cell_type": "markdown", "metadata": {}, - "source": [ - "## Alternative: user-supplied anchors\n", - "\n", - "If you already have a labeled anchor table (columns: your attribute key e.g. `domain_sessionid`, `anchor_ts`, and optionally `label`), use `build_dataset_with_custom_anchors()` instead of the session-anchor method. Everything downstream (`save_to`, `execute`, `to_pandas`) is identical.\n", - "\n", - "```python\n", - "from snowplow_signals import WarehouseTable\n", - "\n", - "bundle = sp_signals.build_dataset_with_custom_anchors(\n", - " attribute_groups=[ecommerce_group],\n", - " anchors_table=WarehouseTable(\n", - " database=\"analytics\",\n", - " schema=\"ml\",\n", - " table=\"my_anchor_events\",\n", - " ),\n", - " anchors_have_label=True,\n", - ")\n", - "```" - ] + "source": "## Alternative: user-supplied anchors\n\nIf you already have a labeled anchor table (columns: your attribute key e.g. `domain_sessionid`, `anchor_ts`, and optionally `label`), use `submit_dataset_run_with_custom_anchors()` instead of the session-anchor method. Everything downstream (polling, preview, `to_pandas`) is identical.\n\n```python\nfrom snowplow_signals import WarehouseTable\n\nrun = sp_signals.submit_dataset_run_with_custom_anchors(\n attribute_groups=[ecommerce_group],\n anchors_table=WarehouseTable(\n database=\"analytics\",\n schema=\"ml\",\n table=\"my_anchor_events\",\n ),\n anchors_have_label=True,\n)\n```\n\n## Inspect the generated SQL\n\nIf you want to review the SQL before execution, use `build_dataset_with_session_anchors()` to generate a `DatasetBundle` and save it to disk:\n\n```python\nbundle = sp_signals.build_dataset_with_session_anchors(\n attribute_groups=[ecommerce_group],\n goal_criteria=action_type_is(\"transaction\"),\n training_span=TrainingSpan(\n start_time=datetime(2024, 1, 1, tzinfo=timezone.utc),\n end_time=datetime(2024, 4, 1, tzinfo=timezone.utc),\n ),\n)\nbundle.save_to(\"./dataset_output\")\n```" } ], "metadata": {