mirror of
https://github.com/datahub-project/datahub.git
synced 2025-07-14 12:41:30 +00:00
42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
from datetime import datetime
|
|
|
|
from airflow import DAG
|
|
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
|
|
|
|
SNOWFLAKE_COST_TABLE = "costs"
|
|
SNOWFLAKE_PROCESSED_TABLE = "processed_costs"
|
|
|
|
|
|
def _fake_snowflake_execute(*args, **kwargs):
|
|
raise ValueError("mocked snowflake execute to not run queries")
|
|
|
|
|
|
with DAG(
|
|
"snowflake_operator",
|
|
start_date=datetime(2023, 1, 1),
|
|
schedule_interval=None,
|
|
catchup=False,
|
|
) as dag:
|
|
# HACK: We don't want to send real requests to Snowflake. As a workaround,
|
|
# we can simply monkey-patch the operator.
|
|
SnowflakeOperator.execute = _fake_snowflake_execute # type: ignore
|
|
|
|
transform_cost_table = SnowflakeOperator(
|
|
snowflake_conn_id="my_snowflake",
|
|
task_id="transform_cost_table",
|
|
sql="""
|
|
CREATE OR REPLACE TABLE {{ params.out_table_name }} AS
|
|
SELECT
|
|
id,
|
|
month,
|
|
total_cost,
|
|
area,
|
|
total_cost / area as cost_per_area
|
|
FROM {{ params.in_table_name }}
|
|
""",
|
|
params={
|
|
"in_table_name": SNOWFLAKE_COST_TABLE,
|
|
"out_table_name": SNOWFLAKE_PROCESSED_TABLE,
|
|
},
|
|
)
|