# Building Dagster-Snowflake Assets

This guide explains how to build and configure Dagster assets that interact with Snowflake tables at Suno. This covers both regular assets (with logic executed by Dagster) and external asset specifications (for tracking tables materialized outside Dagster).

> **Note**: This guide does *not* include dbt assets. For dbt integration, see the [Dagster-DBT guide](DAGSTER_DBT_GUIDE.md).

## Table of Contents

- [Asset Types](#asset-types)
- [Key Concepts](#key-concepts)
- [Assets](#assets)
  - [Partitioning Data](#partitioning-data)
  - [Dependencies](#dependencies)
  - [Scheduling (Automation Conditions)](#scheduling-automation-conditions)
  - [Backfill Policies](#backfill-policies)
  - [Handling Partitions in Materializations](#handling-partitions-in-materializations)
  - [Asset Configuration](#asset-configuration)
- [AssetSpec (External Tables)](#assetspec-external-tables)
  - [When to Use AssetSpec](#when-to-use-assetspec)
  - [Basic AssetSpec Definition](#basic-assetspec-definition)
  - [AssetSpec vs Asset](#assetspec-vs-asset)
  - [AssetSpec with Dependencies](#assetspec-with-dependencies)
- [Sensors for Data Monitoring](#sensors-for-data-monitoring)
  - [Basic Sensor Structure](#basic-sensor-structure)
  - [Integration with Automation Conditions](#integration-with-automation-conditions)
- [Snowflake Warehouse Configuration](#snowflake-warehouse-configuration)
- [Best Practices](#best-practices)
- [Troubleshooting](#troubleshooting)
- [Additional Resources](#additional-resources)

## Asset Types

### Assets (`@dg.asset`)
**Primary approach** - Tables with materialization logic executed directly by Dagster:

| Feature | Description |
|---------|-------------|
| **Custom SQL Logic** | Defined in Python functions with full control |
| **Execution Control** | Complete control over execution, error handling, and optimization |
| **Partition Context** | Direct access to partition context and upstream data |
| **Use Case** | Recommended for new pipelines and complex transformations |
| **Monitoring** | Built-in Dagster monitoring and logging |

### AssetSpec (`dg.AssetSpec`)
**Special case** - External tables tracked in Dagster's dependency graph:

| Feature | Description |
|---------|-------------|
| **External Materialization** | Materialized by Snowflake procedures, tasks, or external ETL |
| **Logic Location** | Used when logic cannot be moved into Dagster |
| **Data Monitoring** | Requires sensors for data monitoring |
| **Use Case** | Primarily for legacy systems and third-party integrations |
| **Dependency Tracking** | Still tracked in Dagster's dependency graph |

## Key Concepts

- **Partitioning**: Time-based data segmentation (hourly/daily) for incremental processing
- **Dependencies**: Asset relationships with partition mapping controls
- **Automation Conditions**: Declarative scheduling that waits for upstream dependencies
- **Backfill Policies**: Strategies for processing historical partition ranges
- **Configuration**: Runtime parameters like warehouse sizing for different workloads

## Assets

Dagster Assets represent nodes in the graph; for our purposes this is primarily one asset per Snowflake table. They are used in the following way:

```python
@dg.asset(
    name="fact_hook_play",
    description="Fact table for hook plays.",
    group_name="hooks",
    partitions_def=dg.HourlyPartitionsDefinition(start_date=FACT_HOOK_PLAY_START_DATE, end_offset=-1),
    deps=[
        dg.AssetDep(stg_web_hook_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_ios_hook_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_android_hook_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
    ],
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
    owners=[Team.DATA_POD.value],
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": FACT_HOOK_PLAY_TABLE_NAME,
        "data_start_date": FACT_HOOK_PLAY_START_DATE.strftime("%Y-%m-%d"),
        "cluster_by": "[p_date, p_hour]",
        "partition_expr": PartitionExpr.HOURLY.value,
    },
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
)
def fact_hook_play(context: dg.AssetExecutionContext, snowflake: SnowflakeResource, config: FactHookPlayConfig) -> dg.MaterializeResult:
    ...
```

We detail some of these concepts in the sections below.

### Partitioning data

```python
@dg.asset(
    ...
    partitions_def=dg.HourlyPartitionsDefinition(start_date=FACT_HOOK_PLAY_START_DATE, end_offset=-1),
)
```

Dagster's definition of partitions runs separate from Snowflake's partitioning. This means that Dagster does not automatically know whether a Snowflake table is partitioned, and similarly, defining a Dagster partition does not ensure that the Snowflake asset is partitioned; in order to ensure that these are handled correctly, the partition mappings must be defined by passing through Dagster partition keys to the SQL query that filters based on those keys.

We primarily use hourly or daily partitions for most Snowflake assets. For hourly partitions, while our Snowflake partitions are defined in two separate columns (`p_date` and `p_hour`), Dagster uses an hourly partition key format of `YYYY-MM-DD-HH:00`.

A typical partition definition takes in a `start_date` and also an `end_offset`, which defines an offset beyond the most recent hour of data. For example in the example for `fact_hook_play` above, an offset of -1 means that this table has a 2-hour SLA; the data from 1-2pm will be materialized during 3-4pm. This has no functional importance in scheduling, but has an importance in freshness checks.

### Dependencies

```python
@dg.asset(
    ...
    deps=[
        dg.AssetDep(stg_web_hook_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_ios_hook_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
        dg.AssetDep(stg_android_hook_play, partition_mapping=dg.TimeWindowPartitionMapping(allow_nonexistent_upstream_partitions=True)),
    ],
)
```

One of the strongest power features of Dagster is robust dependency tracking. Unlike dbt (which automatically tracks table dependencies), Dagster requires the user to manually define upstream asset dependencies. Dependencies rely on partition mappings to know which upstream partitions correspond to which downstream partitions.

In the example above from `fact_hook_play` above, we also see a parameter `allow_nonexistent_upstream_partitions=True`. By default this is `False`, but when intentionally aggregating tables that may have different start dates on partitions, this parameter can be used.

The default partition mapping is an identity mapping - partitions upstream correspond to the same partitions downstream. However, we can use a custom partition mapping if needed. Below is an example from a table that covers sessions data; in this case, because sessions can span hour boundaries, we want to collect two partitions of data from the upstream data source when computing one downstream partition; the code below achieves this.

```python
@dg.asset(
    ...
    deps=[
        dg.AssetDep(app_event, partition_mapping=dg.TimeWindowPartitionMapping(start_offset=0, end_offset=1)),
        dg.AssetDep(rds_video_hook),
        dg.AssetDep(dim_clip),
        dg.AssetDep(dim_user),
    ],
)
```

### Scheduling (Automation conditions)

We typically use Dagster for two things: (1) to materialize the most recent partition of data on a graph and propagate changes downstream, and (2) to backfill past partitions of data. Dagster has a declarative system called Automation Conditions that support both use cases quite well.

```python
@dg.asset(
    ...
    automation_condition=hourly_cron_with_eager_historical_backfill_condition,
)
```

Dagster provides off-the-shelf automations for cron jobs that wait for upstream dependencies to finish materialize. We also define an eager automation condition that detects any upstream data changes and automatically kicks off the requisite downstream data changes based on the partition mappings defined above. We combine the two to define the two automation conditions that are used for most of our tasks. More details are in [`src/utils/automation_conditions.py`](../src/utils/automation_conditions.py).

This repository also contains some legacy Schedules, which are simple cron jobs that execute on a cron schedule. They do not necessarily wait for upstream data to be backfilled, so use of automation conditions are encouraged over cron scheduling where possible.

### Backfill policies

![Dagster materialize partitions UI](images/dagster_materialize_partitions.png)

The Dagster UI allows us to materialize multiple partitions at a time. By default, this will launch one run per partition; however, it is often more efficient to launch backfills for multiple partitions per run, especially when running a large backfill. For these cases, we use a backfill policy that allows us to simultaneously materialize multiple partitions per run:

```python
@dg.asset(
    ...
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24*7*2),
)
```

In the example above, we dictate that the backfills will run two weeks' worth of hourly partitions instead of 336 individual queries. Two weeks is a good guideline, but the actual value used here depends on the query being run. When backfilling the asset, Dagster passes a partition range to the context, which can be extracted and fed into the SQL queries as parameters. This is covered in the next section.

### Handling partitions in materializations

The key to working with partitioned assets is extracting partition information from the context and using it to filter data appropriately. Here's how to handle both single partition and multi-partition (backfill) scenarios:

#### Key Context Methods for Partitions

```python
@dg.asset(partitions_def=dg.HourlyPartitionsDefinition(start_date="2023-01-01"))
def my_asset(context: dg.AssetExecutionContext, snowflake: SnowflakeResource):
    # Single partition key (string format: "2023-01-01-14:00")
    partition_key = context.partition_key

    # Partition time window (datetime objects)
    start_time = context.partition_time_window.start
    end_time = context.partition_time_window.end

    # Check if this is a backfill with multiple partitions
    is_multi_partition_range = context.has_partition_key_range

    # Get upstream asset partition windows
    upstream_window = context.asset_partitions_time_window_for_input("upstream_asset")

    # Use in SQL queries
    query = f"""
    SELECT * FROM events
    WHERE event_timestamp >= '{start_time.strftime('%Y-%m-%d %H:%M:%S')}'
      AND event_timestamp < '{end_time.strftime('%Y-%m-%d %H:%M:%S')}'
    """
```

#### Handling Different Partition Start Dates

When aggregating data from multiple sources that have different start dates, use conditional logic:

```python
@dg.asset
def aggregated_events(context, snowflake):
    partition_end = context.partition_time_window.end

    # Check if data sources have data for this partition
    include_web_events = partition_end.timestamp() >= WEB_EVENTS_START_DATE.timestamp()
    include_mobile_events = partition_end.timestamp() >= MOBILE_EVENTS_START_DATE.timestamp()

    if include_web_events:
        # Process web events
        web_window = context.asset_partitions_time_window_for_input("web_events")
        # ... web processing logic

    if include_mobile_events:
        # Process mobile events
        mobile_window = context.asset_partitions_time_window_for_input("mobile_events")
        # ... mobile processing logic
```

### Asset Configuration

```python
@dg.asset
def my_asset(context, snowflake, config: MyAssetConfig):
    # Config allows runtime parameters like warehouse selection
    with snowflake.get_connection() as conn:
        cursor = conn.cursor()

        # Set warehouse based on config (useful for backfills)
        cursor.execute(f"USE WAREHOUSE {config.warehouse}")

        # Execute queries
        cursor.execute(query)
        rows_affected = cursor.rowcount
        conn.commit()

    return dg.MaterializeResult(
        metadata={"dagster/row_count": rows_affected}
    )
```

## AssetSpec (External Tables)

`AssetSpec` is used for external tables that are materialized outside of Dagster (e.g., by stored procedures, ETL jobs) but need to be tracked in Dagster's asset graph for dependency management.

### When to Use AssetSpec

Use `AssetSpec` when:
- Tables are materialized by external Snowflake procedures/tasks
- You need to track these tables as dependencies for downstream assets
- You want to monitor for new data via sensors
- The materialization logic cannot be moved into Dagster

### Basic AssetSpec Definition

```python
from datetime import datetime
import dagster as dg
from dagster import EnvVar
from src.utils.snowflake.constants import PartitionExpr

DIM_USER_TABLE_NAME = "DIM_USER"
DIM_USER_START_DATE = datetime.strptime('2023-03-26', '%Y-%m-%d')

dim_user = dg.AssetSpec(
    key="dim_user",
    description="Dimension table for user information, partitioned by date user joined.",
    group_name="identity",
    partitions_def=dg.HourlyPartitionsDefinition(start_date=DIM_USER_START_DATE),
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": DIM_USER_TABLE_NAME,
        "partition_expr": PartitionExpr.HOURLY.value,
    }
)
```

### AssetSpec vs Asset

| Aspect | AssetSpec | @asset |
|--------|-----------|---------|
| **Materialization** | External (procedures/tasks) | Dagster-executed |
| **Logic Location** | Outside Dagster | Inside Dagster function |
| **Data Monitoring** | Requires sensors | Optional sensors |
| **Dependency Tracking** | Yes | Yes |
| **Partitioning** | Yes | Yes |
| **Use Case** | Legacy systems, external ETL | New pipelines, custom logic |

### AssetSpec with Dependencies

```python
fact_user_activity = dg.AssetSpec(
    key="fact_user_activity",
    deps=["dim_user", "dim_clip"],  # Depends on both dimension tables
    partitions_def=dg.DailyPartitionsDefinition(start_date=START_DATE),
    automation_condition=dg.AutomationCondition.on_missing(),  # Manual trigger only
    metadata={
        "database": EnvVar("SNOWFLAKE_DB").get_value(),
        "schema": EnvVar("SNOWFLAKE_SCHEMA").get_value(),
        "table_name": "FACT_USER_ACTIVITY",
        "partition_expr": PartitionExpr.DAILY.value,
    }
)
```

## Sensors for Data Monitoring

Sensors monitor Snowflake tables for new data and trigger asset materializations. They are primarily used with AssetSpec assets to detect when external processes have materialized new data.

### Basic Sensor Structure

```python
@dg.sensor(
    name="snowflake_new_data_sensor__dim_user",
    description="Sensor that monitors dim_user for new data",
    minimum_interval_seconds=300,  # Check every 5 minutes
    metadata={
        "expected_data_cadence": "hourly"
    }
)
def dim_user_new_data_sensor(
    context: dg.SensorEvaluationContext,
    snowflake: SnowflakeResource
) -> dg.SensorResult:
    """Sensor that monitors dim_user for new data and triggers materialization."""
    logger = context.log

    current_hour = datetime.now().replace(minute=0, second=0, microsecond=0)
    previous_hour = current_hour - timedelta(hours=1)
    partition_key = previous_hour.strftime("%Y-%m-%d-%H:00")

    # Handle first run and cursor logic
    if context.is_first_tick_since_sensor_start:
        last_processed_partition = MIN_TIMESTAMP.strftime("%Y-%m-%d-%H:00")
    elif context.cursor:
        last_processed_partition = context.cursor
        if last_processed_partition == partition_key:
            return dg.SensorResult(skip_reason="Already processed this partition")
    else:
        last_processed_partition = MIN_TIMESTAMP.strftime("%Y-%m-%d-%H:00")

    # Query Snowflake for new data
    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        query = f"""
            SELECT COUNT(*)
            FROM {dim_user.key.to_user_string()}
            WHERE DATE_TRUNC('hour', created_at) = '{partition_key[:13]}:00:00'
        """
        cursor.execute(query)
        result = cursor.fetchone()
        row_count = result[0] if result else 0

    # Emit materialization event if new data found
    if row_count > 0:
        return dg.SensorResult(
            cursor=partition_key,
            asset_events=[
                dg.AssetMaterialization(
                    asset_key=dim_user.key,
                    partition=partition_key,
                    metadata={
                        "source": "snowflake_new_data_sensor__dim_user",
                        "dagster/row_count": row_count,
                    }
                )
            ]
        )
    else:
        return dg.SensorResult(skip_reason="No new data found")
```

### Integration with Automation Conditions

Sensors work well with automation conditions to create sophisticated workflows:

```python
# Asset that waits for sensor-detected data AND upstream dependencies
@dg.asset(
    automation_condition=(
        dg.AutomationCondition.any_deps_match(lambda asset: asset.key == "dim_user") &
        dg.AutomationCondition.all_deps_updated_since_cron("0 * * * *")
    )
)
def downstream_asset(context, snowflake, dim_user):
    # This will run when dim_user gets new data via sensor
    # AND all other dependencies are updated within the last hour
    pass
```

## Snowflake Warehouse Configuration

### Available Warehouses

| Warehouse | Use Case | Performance Level | Recommended For |
|-----------|----------|-------------------|-----------------|
| `DBT_DEV_MEDIUM` | Development and light processing | Medium | Testing, small datasets |
| `FACT_HOOK_PLAY_LARGE` | Heavy hook-related processing | Large | Hook play aggregations |
| `HOOK_SESSION_X_SMALL` | Hook session analysis | Large | Session-based analytics |
| `DIM_HOOK_HOURLY_LARGE` | Dimension table processing | Large | Dimension table updates |
| `ANALYTICS_LARGE` | General analytics workloads | Large | Complex analytical queries |
| `ANALYTICS_XLARGE` | Very heavy processing | X-Large | Large backfills, complex aggregations |

> **Note**: More warehouses are available - refer to Snowflake to see all options.

### Warehouse Selection Guidelines

```python
# For development and testing
@dg.asset
def dev_asset(context, snowflake, config):
    warehouse = "DBT_DEV_MEDIUM"
    # ... implementation

# For production workloads
@dg.asset
def prod_asset(context, snowflake, config):
    warehouse = config.warehouse or "ANALYTICS_LARGE"
    # ... implementation

# For very heavy processing (backfills, large aggregations)
@dg.asset(
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=1)
)
def heavy_asset(context, snowflake, config):
    warehouse = "ANALYTICS_XLARGE"
    # ... implementation
```

### Dynamic Warehouse Selection

```python
from enum import Enum

class Warehouse(Enum):
    DEV = "DBT_DEV_MEDIUM"
    MEDIUM = "ANALYTICS_MEDIUM"
    LARGE = "ANALYTICS_LARGE"
    XLARGE = "ANALYTICS_XLARGE"

@dg.asset
def adaptive_asset(context, snowflake, config):
    is_multi_partition_range = context.has_partition_key_range
    # Choose warehouse based on partition count or data volume
    if is_multi_partition_range:
        # Backfill scenario - use larger warehouse
        warehouse = Warehouse.XLARGE.value
    else:
        # Single partition - use standard warehouse
        warehouse = Warehouse.LARGE.value

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(f"USE WAREHOUSE {warehouse}")
        # ... rest of implementation
```

### Warehouse Performance Tips

- **Use appropriate sizing**: Start with medium, scale up based on query performance
- **Monitor usage**: Check warehouse utilization in Snowflake UI
- **Batch operations**: Use larger warehouses for backfills, smaller for incremental updates
- **Auto-suspend**: Configure warehouses to auto-suspend when idle
- **Query optimization**: Optimize queries before scaling up warehouse size

## Best Practices

### 1. Asset Organization

```python
# Group related assets for better UI organization
group_name="identity"     # For user-related tables
group_name="content"      # For clip/song-related tables
group_name="engagement"   # For play/interaction tables
group_name="hooks"        # For hooks-related tables
```

### 2. Naming Conventions

```python
# Use consistent prefixes based on data layer
dim_user          # Dimension tables
fact_hook_play    # Fact tables
agg_user_metrics  # Aggregated tables
stg_events        # Staging tables
```

### 3. Error Handling

```python
@dg.asset
def robust_asset(context, snowflake):
    try:
        with snowflake.get_connection() as conn:
            cursor = conn.cursor()
            # Database operations
            cursor.execute(query)
            conn.commit()

        return dg.MaterializeResult(metadata={"status": "success"})

    except Exception as e:
        context.log.error(f"Asset materialization failed: {str(e)}")
        # Optionally re-raise or return failure metadata
        raise
```

### 4. Performance Optimization

```python
@dg.asset(
    metadata={
        "warehouse": Warehouse.LARGE.value,  # Use appropriate warehouse size
    },
    backfill_policy=dg.BackfillPolicy.multi_run(max_partitions_per_run=24)  # Batch backfills
)
def optimized_asset(context, snowflake, config):
    # Use config for dynamic warehouse sizing
    warehouse = config.warehouse if hasattr(config, 'warehouse') else Warehouse.MEDIUM.value

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(f"USE WAREHOUSE {warehouse}")
        # ... rest of logic
```

### 5. Error Handling and Resilience

```python
@dg.asset
def resilient_asset(context, snowflake):
    max_retries = 3
    retry_count = 0

    while retry_count < max_retries:
        try:
            with snowflake.get_connection() as conn:
                cursor = conn.cursor()
                # Your database operations here
                cursor.execute(query)
                conn.commit()

            return dg.MaterializeResult(
                metadata={"status": "success", "retry_count": retry_count}
            )

        except Exception as e:
            retry_count += 1
            context.log.warning(f"Attempt {retry_count} failed: {str(e)}")

            if retry_count >= max_retries:
                context.log.error(f"Asset failed after {max_retries} attempts")
                raise
            else:
                time.sleep(2 ** retry_count)  # Exponential backoff
```

### 6. Data Quality and Validation

```python
@dg.asset
def validated_asset(context, snowflake):
    with snowflake.get_connection() as conn:
        cursor = conn.cursor()

        # Execute main query
        cursor.execute(main_query)
        rows_affected = cursor.rowcount

        # Data quality checks
        cursor.execute("SELECT COUNT(*) FROM my_table WHERE p_date = ?", [context.partition_key])
        row_count = cursor.fetchone()[0]

        # Validate expected data volume
        if row_count < expected_min_rows:
            raise ValueError(f"Data quality check failed: {row_count} rows < {expected_min_rows}")

        return dg.MaterializeResult(
            metadata={
                "dagster/row_count": row_count,
                "data_quality": "passed"
            }
        )
```

### 7. Monitoring and Observability

```python
@dg.asset
def monitored_asset(context, snowflake):
    start_time = time.time()

    with snowflake.get_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(query)
        rows_affected = cursor.rowcount

    execution_time = time.time() - start_time

    return dg.MaterializeResult(
        metadata={
            "dagster/row_count": rows_affected,
            "execution_time_seconds": execution_time,
            "warehouse_used": context.resources.snowflake.warehouse,
            "partition_key": context.partition_key
        }
    )
```

## Troubleshooting

### Common Issues

#### Connection Failures
- **Issue**: Snowflake connection timeouts or failures
- **Solution**: Check network connectivity, credentials, and warehouse availability
- **Prevention**: Implement retry logic with exponential backoff

#### Partition Data Issues
- **Issue**: Missing or incorrect partition data
- **Solution**: Verify partition filtering logic and upstream dependencies
- **Debug**: Use `context.log.info()` to log partition keys and data counts

#### Warehouse Performance
- **Issue**: Queries running slowly or timing out
- **Solution**: Scale up warehouse size or optimize query logic
- **Monitoring**: Check Snowflake query history for performance insights

#### Backfill Failures
- **Issue**: Backfills failing due to resource constraints
- **Solution**: Reduce `max_partitions_per_run` or use larger warehouse
- **Alternative**: Process backfills in smaller batches

### Debugging Tips

1. **Enable detailed logging** in asset functions
2. **Check Dagster logs** for detailed error messages
3. **Monitor Snowflake query history** for performance issues
4. **Use partition-specific debugging** to isolate issues
5. **Test with single partitions** before running full backfills

### Performance Monitoring

```python
@dg.asset
def performance_monitored_asset(context, snowflake):
    # Log performance metrics
    context.log.info(f"Starting materialization for partition: {context.partition_key}")

    start_time = time.time()
    # ... your logic here
    execution_time = time.time() - start_time

    context.log.info(f"Materialization completed in {execution_time:.2f} seconds")

    return dg.MaterializeResult(
        metadata={
            "execution_time_seconds": execution_time,
            "partition_key": context.partition_key
        }
    )
```

## Additional Resources

- [Dagster Documentation](https://docs.dagster.io/)
- [Snowflake Documentation](https://docs.snowflake.com/)
- [Snowflake Python Connector](https://docs.snowflake.com/en/user-guide/python-connector.html)
- Internal Slack channels: `#data-engineering`, `#snowflake-help`
- Snowflake UI: Monitor warehouse usage and query performance

---

This guide provides a comprehensive foundation for building both regular Assets and AssetSpec external tables in Dagster, with proper partitioning, automation, and monitoring strategies. For questions or improvements to this guide, please reach out to the Data Engineering team.