Workflow orchestration for Snowflake

1from prefect import flow, task
2from prefect_snowflake import SnowflakeConnector
3
4
5@task(retries=3, retry_delay_seconds=30)
6def load_raw_events(block_name: str) -> None:
7    with SnowflakeConnector.load(block_name) as connector:
8        connector.execute(
9            """
10            COPY INTO raw.events
11            FROM @events_stage
12            FILE_FORMAT = (TYPE = 'JSON')
13            """
14        )
15
16
17@task
18def transform_daily(block_name: str) -> None:
19    with SnowflakeConnector.load(block_name) as connector:
20        connector.execute(
21            """
22            INSERT INTO analytics.daily_events
23            SELECT event_date, event_type, COUNT(*) AS count
24            FROM raw.events
25            WHERE event_date = CURRENT_DATE
26            GROUP BY 1, 2
27            """
28        )
29
30
31@flow(name="snowflake-daily-etl")
32def snowflake_daily_etl(block_name: str = "prod-warehouse") -> None:
33    load_raw_events(block_name)
34    transform_daily(block_name)
35
36
37if __name__ == "__main__":
38    snowflake_daily_etl()