Define Assets for the ETL Pipeline

Now, we'll define a series of assets for the ETL pipeline. Assets must be organized properly so they can be discovered by Dagster.

Create the assets module: Create a file named assets.py in the defs/ folder and add the following code to define the pipeline:

import pandas as pd
from pathlib import Path
from dagster import asset

@asset(required_resource_keys={"teradata"})
def read_csv_file(context):
    csv_path = Path(__file__).parent.parent.parent.parent / "data" / "sample_data.csv"
    df = pd.read_csv(csv_path)
    context.log.info(df)
    return df

@asset(required_resource_keys={"teradata"})
def drop_table(context):
    try:
        result = context.resources.teradata.drop_table(["dagster_pipeline_db.tmp_table"])
        context.log.info(result)
    except Exception as e:
        context.log.warning(f"Drop table warning (may not exist): {e}")

@asset(required_resource_keys={"teradata"})
def create_table(context, drop_table):
    try:
        result = context.resources.teradata.execute_query('''CREATE TABLE dagster_pipeline_db.tmp_table (
                                                                id INTEGER,
                                                                name VARCHAR(50),
                                                                age INTEGER,
                                                                city VARCHAR(50));''')
        context.log.info(result)
    except Exception as e:
        context.log.error(f"Failed to create table: {e}")
        raise

@asset(required_resource_keys={"teradata"}, deps=[read_csv_file])
def insert_rows(context, create_table, read_csv_file):
    try:
        data_tuples = [tuple(row) for row in read_csv_file.to_numpy()]
        for row in data_tuples:
            result = context.resources.teradata.execute_query(
                f"INSERT INTO dagster_pipeline_db.tmp_table (id, name, age, city) VALUES ({row[0]}, '{row[1]}', {row[2]}, '{row[3]}');"
            )
            context.log.info(result)
    except Exception as e:
        context.log.error(f"Failed to insert rows: {e}")
        raise

@asset(required_resource_keys={"teradata"})
def read_table(context, insert_rows):
    try:
        result = context.resources.teradata.execute_query("select * from dagster_pipeline_db.tmp_table;", True)
        context.log.info(result)
    except Exception as e:
        context.log.error(f"Failed to read table: {e}")
        raise

This Dagster pipeline defines a series of assets that interact with Teradata. It starts by reading data from a CSV file, then drops and recreates a table in Teradata. After that, it inserts rows from the CSV into the table and finally retrieves the data from the table.