Dagster + Snowflake: Cortexによる本番AIパイプラインの構築

Meet Compass — Dagster’s new AI data analyst for Slack. Turn questions into trusted insights, instantly. Try Compass now →Discover What assets do best, an animated, narrated story about how data assets work together. Watch now →

Try Dagster+Sign InSnowflake handles AI compute while Dagster handles orchestration, observability, and the operational patterns that turn AI experiments into reliable production pipelines.

Snowflake has evolved into an AI compute platform. Cortex AI lets you run sentiment analysis on millions of rows, extract entities from unstructured text, and generate intelligent summaries directly in SQL without moving your data.

However, one does not simply integrate AI into their data platform.

When do those AI queries run? What happens when they fail? How do you track costs? Who coordinates the steps between raw data and actionable intelligence?

Snowflake handles AI compute. Dagster handles orchestration, observability, and the operational patterns that turn AI experiments into reliable production pipelines.

This post walks through a real project: a Hacker News Intelligence Pipeline that demonstrates how Dagster and Snowflake work together. The pipeline includes authentication, incremental processing, cost tracking, and patterns your team can adopt today.

Dagster + Snowflake work well together because each platform does what it does best:

The separation is clean: Snowflake executes, Dagster orchestrates.

Hacker News Intelligence Pipeline

Our demo project scrapes Hacker News stories, enriches them with Cortex AI, and produces daily intelligence summaries. The pipeline covers the patterns needed for production AI workloads:

This end-to-end project showcases the power of both platforms and their complementary nature.

Pattern 1: Cortex AI Orchestration

Cortex AI functions like AI_CLASSIFY, COMPLETE, and AI_AGG run inference at scale. But production requires more than the function itself. You need scheduling, dependency tracking, and monitoring around it. LLM usage can be expensive if you are not monitoring it and only executing when needed. When you wrap Snowflake SQL inside Dagster assets, you get data lineage for the transformation, detailed run history, and versioning.

Dagster orchestrates Cortex AI using Assets like this in Python:

@dg.asset( deps=["scraped_story_content"], compute_kind="snowflake-cortex" ) def story_sentiment_analysis( context: dg.AssetExecutionContext, snowflake: SnowflakeResource ) -> dg.MaterializeResult: """ Dagster orchestrates WHEN this runs and TRACKS the execution. Snowflake executes the AI inference. """ with snowflake.get_connection() as conn: cursor = conn.cursor() # Snowflake runs AI_CLASSIFY - no data movement cursor.execute(f""" UPDATE {schema}.stories SET sentiment = AI_CLASSIFY( content, ['positive', 'neutral', 'negative'] ):label::STRING WHERE sentiment IS NULL """) # Get processing stats for observability cursor.execute(f""" SELECT COUNT(*) as total, COUNT_IF(sentiment = 'positive') as positive, COUNT_IF(sentiment = 'negative') as negative FROM {schema}.stories """) stats = cursor.fetchone() # Dagster captures metadata for lineage and monitoring return dg.MaterializeResult( metadata={ "stories_processed": dg.MetadataValue.int(stats[0]), "positive_count": dg.MetadataValue.int(stats[1]), "negative_count": dg.MetadataValue.int(stats[2]), "sentiment_ratio": dg.MetadataValue.float( stats[1] / stats[0] if stats[0] > 0 else 0 ) } )What's happening here:

Dagster schedules when sentiment analysis runs (after ingestion completes)

Dagster tracks dependencies (waits for scraped_story_content)

Snowflake executes AI_CLASSIFY directly on the data without extraction

Dagster captures metadata for observability and cost tracking

This pattern scales to any Cortex function. Entity extraction with COMPLETE:

@dg.asset(deps=["story_entity_extraction"]) def daily_intelligence_summary(context, snowflake): """Generate AI-powered daily briefing.""" with snowflake.get_connection() as conn: cursor = conn.cursor() cursor.execute(f""" INSERT INTO {schema}.daily_summaries SELECT CURRENT_DATE() as summary_date, AI_AGG( content, 'Synthesize these Hacker News stories into a 3-paragraph tech intelligence briefing. Include: (1) dominant themes, (2) notable launches or announcements, (3) emerging trends to watch. Be specific with names and details.' ) as intelligence_summary, COUNT(*) as stories_analyzed FROM {schema}.stories WHERE DATE(posted_at) = CURRENT_DATE() """)Cortex runs the inference. Dagster ensures it runs reliably, in the correct order, with full observability.

Pattern 2: Incremental Processing at Scale

Processing your entire dataset every run is expensive. Leveraging Dagster partitions and Snowflake MERGE patterns enables efficient incremental processing.

Define daily partitions daily_partitions = dg.DailyPartitionsDefinition( start_date="2024-01-01", timezone="America/New_York" ) @dg.asset