diff --git a/quickstarts/images/dagster/dagster-teradata1.png b/quickstarts/images/dagster/dagster-teradata1.png index 67c6d27e19d..9f03ad0950e 100644 Binary files a/quickstarts/images/dagster/dagster-teradata1.png and b/quickstarts/images/dagster/dagster-teradata1.png differ diff --git a/quickstarts/images/dagster/dagster-teradata2.png b/quickstarts/images/dagster/dagster-teradata2.png index 2160a0b633f..6422e15c53b 100644 Binary files a/quickstarts/images/dagster/dagster-teradata2.png and b/quickstarts/images/dagster/dagster-teradata2.png differ diff --git a/quickstarts/images/dagster/dagster-teradata3.png b/quickstarts/images/dagster/dagster-teradata3.png new file mode 100644 index 00000000000..ea83caca5a6 Binary files /dev/null and b/quickstarts/images/dagster/dagster-teradata3.png differ diff --git a/quickstarts/manage-data/Pushdown-feature.md b/quickstarts/manage-data/Pushdown-feature.md deleted file mode 100644 index dc3781cf037..00000000000 --- a/quickstarts/manage-data/Pushdown-feature.md +++ /dev/null @@ -1,150 +0,0 @@ -# Pushdown Functionality -The Teradata connector supports pushdown for different operations. - -* Join pushdown - - - Cost-based join pushdown - -* Limit pushdown - -* Top N pushdown - -* Projection pushdown - -* Predicate pushdown - -* Aggregate pushdown - -## Join pushdown -Join pushdown allows the Teradata connector to delegate the table join operation to the Teradata. This can result in performance gains and allows Trino to perform the remaining query processing on a smaller amount of data. - -## Cost-based join pushdown -The Teradata connector supports cost-based Join pushdown to make intelligent decisions about whether to push down a join operation to the Teradata. This is based on a catalog configuration property. - -When cost-based join pushdown is enabled, the connector only pushes down join operations if statistics of joining tables are available incase of join pushdown strategy is ```AUTOMATIC```. - -The following table describes catalog configuration properties for join pushdown. - -| Property name | Description | Default value | -|-------------------------|-----------------------------------------------------------------------------|---------------| -| join-pushdown.enabled | Enable join pushdown. `join_pushdown_enabled` is the equivalent catalog session property. | true | -| join-pushdown.strategy | Strategy used to evaluate whether join operations are pushed down. Set to `AUTOMATIC` to enable cost-based join pushdown, or `EAGER` to pushdown joins. Note that `EAGER` can push down joins even when table statistics are unavailable. | AUTOMATIC | - -The connector does not support pushdown of range predicates, such as ```>```, ```<```, or ```BETWEEN```, on columns with character string types like ```CHAR``` or ```VARCHAR```. Equality predicates, such as ```IN``` or ```=```, and inequality predicates, such as ```!=``` on columns with textual types are pushed down. This ensures correctness of results since the remote data source may sort strings differently than Trino. - - -## Limit pushdown -A LIMIT reduces the number of returned rows for a SQL SELECT statement. Limit pushdown enables connector to push processing of such queries of unsorted records to Teradata. A pushdown of this clause can improve the performance of the query and significantly reduce the amount of data transferred from Teradata to Trino. - -Teradata converts Trino LIMIT to Teradata equivalent TOP clause. - -```SELECT id, name FROM teradata.public.company LIMIT 5;``` - -Trino query will be converted to - -```SELECT TOP 5 id, name FROM teradata.public.company;``` - -## TOP N pushdown -The combination of a LIMIT with ORDER BY clause creates a small set of records to return out of a large sorted dataset. The pushdown for a query involving LIMIT with ORDER BY is called a Top N pushdown. It enabled the Teradata connector to push the processing of Queries involved in LIMIT with ORDER BY clause to Teradata database, and therefore significantly reduces the amount of data transferred to Trino. - -```SELECT id, name FROM teradata.public.company ORDER BY id LIMIT 5;``` - -Trino query will be converted to Teradata supported syntax - -```SELECT TOP 5 id, name from (SELECT id, name FROM teradata.public.company) as t ORDER BY id;``` - -## Projection pushdown -Projection pushdown optimizes column-based filtering. It uses the column specified in the SELECT clause and other parts of the query to limit access to these columns. The processing is pushed down to the Teradata database by Teradata connector and then Teradata reads and returns only necessary columns. - -Teradata connector inherits the TRINO JDBC framework implementation. Teradata connector doesn’t have any custom implementation with respect to Projection Pushdown. - -## Predicate pushdown -Predicate pushdown optimizes row-based filtering. It uses filtering specified in WHERE clause to omit unnecessary rows. The processing is pushed down to Teradata database by the Teradata connector and Teradata only reads and returns the filter matched rows only. - -The predicate pushdown supports SQL statements with following arithmetic, boolean, comparision operators and functions. - -* `=` - -* `<>` - -* `<` - -* `<=` - -* `>` - -* `>=` - -* `+` - -* `-` - -* `*` - -* `/` - -* `MOD` - -* `-Value` - -* `IS NOT NULL` - -* `NOT` - -* `IS NULL` - -* `NULLIF` - -* `IN` - -* `LIKE` - -* `LIKE ESCAPE` - -## Aggregate pushdown -Teradata supports Aggregation pushdown for below specified cases. - -### Numeric Data -* CountAll - -* Count - -* Count Distinct - -* Min - -* Max - -* Sum - -* Avg Float - -* Avg Decimal - -* Avg Bigint - -* Statistical Aggregation - - - StddevSamp - - - StddevPop - - - VarianceSamp - - - VariancePop - -* Correlation and Regression Aggregation - - - CovarianceSamp - - - CovariancePop - - - Corr - - - RegrIntercept - - - RegrSlope - -* String Data - -* No Aggregation Pushdown for CHAR and VARCHAR type data diff --git a/quickstarts/manage-data/Trino Installation & Setup.md b/quickstarts/manage-data/Trino Installation & Setup.md deleted file mode 100644 index 29e2b4d256a..00000000000 --- a/quickstarts/manage-data/Trino Installation & Setup.md +++ /dev/null @@ -1,100 +0,0 @@ -# Getting Started with Trino CLI: Installation and Setup -1. Install Eclipse Termurin 24 - -2. Download the trino server file using the below command: - -``` -wget https://repo1.maven.org/maven2/io/trino/trino-server/474/trino-server-474.tar.gz -``` - -3. Extract the contents of the file using below command - -``` -tar -xvzf trino-server-474.tar.gz -``` - -4. create etc directory in trino-server-474 - -5. create below mentioned files under etc directory - -## config.properties - -``` -coordinator=true -node-scheduler.include-coordinator=true -http-server.http.port=8080 -discovery.uri=http://hostname:8080 -``` - -## jvm.config - -``` --server --Xmx16G --XX:InitialRAMPercentage=80 --XX:MaxRAMPercentage=80 --XX:G1HeapRegionSize=32M --XX:+ExplicitGCInvokesConcurrent --XX:+ExitOnOutOfMemoryError --XX:+HeapDumpOnOutOfMemoryError --XX:-OmitStackTraceInFastThrow --XX:ReservedCodeCacheSize=512M --XX:PerMethodRecompilationCutoff=10000 --XX:PerBytecodeRecompilationCutoff=10000 --Djdk.attach.allowAttachSelf=true --Djdk.nio.maxCachedBufferSize=2000000 --Dfile.encoding=UTF-8 -# Allow loading dynamic agent used by JOL --XX:+EnableDynamicAgentLoading -``` - -## node.properties - -``` -node.environment=production -node.id=ffffffff-ffff-ffff-ffff-ffffffffffff -node.data-dir=/var/trino/data -``` - -## catalog directory - -create a file named teradata.properties - - - -## Teradata connector properties - -``` -connector.name=teradata -connection-url=jdbc:teradata://hostname/ -connection-user=*** -connection-password=*** -``` - -6. Download trino-teradata zip file - -7. Extract the zip file and place the folder: **trino-teradata-474** in **plugin** directory of **trino-server-474** - -The structure should be as follows: - -trino-server-474 → plugin → trino-teradata-474 - -8. Download the jar file from the given link in the trino-server-474 directory : - -* https://repo1.maven.org/maven2/io/trino/trino-cli/474/trino-cli-474-executable.jar - -* Make jar as executable by running ```chmod +x``` - -## Start Trino Server -The installation provides a ```bin/launcher``` script that can be used manually or as a daemon startup script. - -``` -bin/launcher start -``` - -## Start Trino CLI -Start Trino CLI using - -``` -java -jar trino-cli-*-executable.jar --version -``` diff --git a/quickstarts/manage-data/use-dagster-with-teradata-vantage.md b/quickstarts/manage-data/use-dagster-with-teradata-vantage.md deleted file mode 100644 index e3c26e15545..00000000000 --- a/quickstarts/manage-data/use-dagster-with-teradata-vantage.md +++ /dev/null @@ -1,301 +0,0 @@ ---- -id: use-dagster-with-teradata-vantage -sidebar_position: 4.5 -author: Mohan Talla, Daniel Herrera -email: developer.relations@teradata.com -page_last_update: 2026-06-23 -description: Use dagster-teradata with Teradata Vantage. -keywords: [dagster, dagster-teradata, data warehouses, compute storage separation, teradata, vantage, cloud data platform, object storage, business intelligence, enterprise analytics, elt] ---- - -import TrialDocsNote from '../_partials/teradata_trial.mdx' -import InstallTabs from '../_partials/tabsDBT.mdx' - -# dagster-teradata with Teradata Vantage - -This guide walks you through integrating Dagster with Teradata Vantage to create and manage ETL pipelines. It provides step-by-step instructions for installing and configuring the necessary packages, setting up a Dagster project, and implementing a pipeline that interacts with Teradata Vantage. - -## Dagster - -* Dagster is a data orchestrator built for data engineers, with integrated lineage, observability, a declarative programming model and best-in-class testability. -* Data pipelines are automated workflows that ingest raw data, process it through various transformations (such as cleaning and structuring), and produce a final, usable format—much like an assembly line for data. -* Dagster orchestrates this process by defining each stage of the pipeline, ensuring tasks execute in the correct sequence and at scheduled intervals. It provides a structured way to manage dependencies, track execution, and maintain reliable data workflows. -* Dagster orchestrates dbt alongside other technologies. Dagster's asset-oriented approach allows Dagster to understand dbt at the level of individual dbt models. - - -## Prerequisites - -* Access to a Teradata Vantage instance (Teradata Cloud, Teradata Factory, or Teradata Trial). - - - -* Python **3.9** or higher, Python **3.12** is recommended. -* pip -* A Teradata database where you have CREATE TABLE privileges. You can create one by running: - - ```sql - CREATE DATABASE dagster_pipeline_db AS PERMANENT = 100e6; - ``` - -## Setting Up a Virtual Enviroment - -A virtual environment is recommended to isolate project dependencies and avoid conflicts with system-wide Python packages. Here's how to set it up: - - - -## Install dagster and dagster-teradata - -With your virtual environment active, the next step is to install dagster and the Teradata provider package (dagster-teradata) to interact with Teradata Vantage. - -1. Install the Required Packages: - - ```bash - pip install dagster dagster-webserver dagster-teradata - ``` - -2. Note about Optional Dependencies: - - a) `dagster-teradata` relies on dagster-aws for ingesting data from an S3 bucket into Teradata Vantage. Since `dagster-aws` is an optional dependency, users can install it by running: - - ```bash - pip install dagster-teradata[aws] - ``` - b) `dagster-teradata` also relies on `dagster-azure` for ingesting data from an Azure Blob Storage container into Teradata Vantage. To install this dependency, run: - - ```bash - pip install dagster-teradata[azure] - ``` - -3. Verify the Installation: - - To confirm that Dagster is correctly installed, run: - ```bash - dagster --version - ``` - If installed correctly, it should show the version of Dagster. - - -## Initialize a Dagster Project - -Now that you have the necessary packages installed, the next step is to create a new Dagster project. - -### Scaffold a New Dagster Project - -Run the following command: - -```bash -dagster project scaffold --name dagster-quickstart - ``` -This command will create a new project named dagster-quickstart. It will automatically generate the following directory structure: - -```bash -dagster-quickstart -│ pyproject.toml -│ README.md -│ setup.cfg -│ setup.py -│ -├───dagster_quickstart -│ assets.py -│ definitions.py -│ __init__.py -│ -└───dagster_quickstart_tests - test_assets.py - __init__.py - ``` - -Refer [here](https://docs.dagster.io/guides/build/projects/dagster-project-file-reference) to know more above this directory structure - -## Create Sample Data - -To simulate an ETL pipeline, create a CSV file with sample data that your pipeline will process. - -**Create the CSV File:** Inside the dagster_quickstart/data/ directory, create a file named sample_data.csv with the following content: - -```bash -id,name,age,city -1,Alice,28,New York -2,Bob,35,San Francisco -3,Charlie,42,Chicago -4,Diana,31,Los Angeles - ``` -This file represents sample data that will be used as input for your ETL pipeline. - -## Create a Database for the Pipeline - -Before defining assets, create a database where the pipeline can create and drop tables: - -```sql -CREATE DATABASE dagster_pipeline_db AS PERMANENT = 100e6; -``` - -Update the database name (`dagster_pipeline_db`) in your environment variables and asset code to match your chosen database name. - -## Define Assets for the ETL Pipeline - -Now, we'll define a series of assets for the ETL pipeline inside the assets.py file. - -Edit the assets.py File: Open the dagster_quickstart/assets.py file and add the following code to define the pipeline: - -```python -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 / "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. - -## Define the Pipeline Definitions - -The next step is to configure the pipeline by defining the necessary resources and jobs. - -**Edit the definitions.py File**: Open dagster_quickstart/definitions.py and define your Dagster pipeline as follows: - -```python -from dagster import EnvVar, Definitions -from dagster_teradata import TeradataResource - -from .assets import read_csv_file, read_table, create_table, drop_table, insert_rows - -# Define the pipeline and resources -defs = Definitions( - assets=[read_csv_file, read_table, create_table, drop_table, insert_rows], - resources={ - "teradata": TeradataResource( - host=EnvVar("TERADATA_HOST"), - user=EnvVar("TERADATA_USER"), - password=EnvVar("TERADATA_PASSWORD"), - database=EnvVar("TERADATA_DATABASE"), - ) - } -) -``` - -This code sets up a Dagster project that interacts with Teradata by defining assets and resources - -1. It imports necessary modules, including pandas, Dagster, and dagster-teradata. -2. It imports asset functions (read_csv_file, read_table, create_table, drop_table, insert_rows) from the assets.py module. -3. It registers these assets with Dagster using Definitions, allowing Dagster to track and execute them. -4. It defines a Teradata resource (TeradataResource) that reads database connection details from environment variables (TERADATA_HOST, TERADATA_USER, TERADATA_PASSWORD, TERADATA_DATABASE). - -## Running the Pipeline - -After setting up the project, you can now run your Dagster pipeline: - -1. **Start the Dagster Dev Server:** In your terminal, navigate to the root directory of your project and run: -dagster dev -After executing the command dagster dev, the Dagster logs will be displayed directly in the terminal. Any errors encountered during startup will also be logged here. Once you see a message similar to: - ```bash - 2025-02-04 09:15:46 +0530 - dagster-webserver - INFO - Serving dagster-webserver on http://127.0.0.1:3000 in process 32564, - ``` - It indicates that the Dagster web server is running successfully. At this point, you can proceed to the next step. -
-2. **Access the Dagster UI:** Open a web browser and navigate to http://127.0.0.1:3000. This will open the Dagster UI where we can manage and monitor your pipelines. -
- ![dagster-teradata1.png](../images/dagster/dagster-teradata1.png) -
-3. **Run the Pipeline:** -* In the top navigation of the Dagster UI, click on Assets > View global asset lineage. -* Click Materialize to execute the pipeline. -* In the popup window, click View to see the details of the pipeline run. -
- ![dagster-teradata2.png](../images/dagster/dagster-teradata2.png) -
- -4. **Monitor the Run:** The Dagster UI allows you to visualize the pipeline's progress, view logs, and inspect the status of each step. You can switch between different views to see the execution logs and metadata for each asset. - -## Below are some of the operations provided by the TeradataResource: - -### 1. Execute a Query (`execute_query`) - -This operation executes a SQL query within Teradata Vantage. - -**Args:** -- `sql` (str) – The query to be executed. -- `fetch_results` (bool, optional) – If True, fetch the query results. Defaults to False. -- `single_result_row` (bool, optional) – If True, return only the first row of the result set. Effective only if `fetch_results` is True. Defaults to False. - -### 2. Execute Multiple Queries (`execute_queries`) - -This operation executes a series of SQL queries within Teradata Vantage. - -**Args:** -- `sql_queries` (Sequence[str]) – List of queries to be executed in series. -- `fetch_results` (bool, optional) – If True, fetch the query results. Defaults to False. -- `single_result_row` (bool, optional) – If True, return only the first row of the result set. Effective only if `fetch_results` is True. Defaults to False. - -### 3. Drop a Database (`drop_database`) - -This operation drops one or more databases from Teradata Vantage. - -**Args:** -- `databases` (Union[str, Sequence[str]]) – Database name or list of database names to drop. - -### 4. Drop a Table (`drop_table`) - -This operation drops one or more tables from Teradata Vantage. - -**Args:** -- `tables` (Union[str, Sequence[str]]) – Table name or list of table names to drop. - - -## Summary -This guide provides a step-by-step approach to integrating Dagster with Teradata Vantage for building ETL pipelines. - -## Further reading -* https://docs.dagster.io/ -* https://docs.dagster.io/getting-started/quickstart -* https://docs.dagster.io/getting-started/installation -* https://docs.dagster.io/etl-pipeline-tutorial/ diff --git a/quickstarts/manage-data/use-dagster-with-teradata.md b/quickstarts/manage-data/use-dagster-with-teradata.md new file mode 100644 index 00000000000..ad5fa683ab8 --- /dev/null +++ b/quickstarts/manage-data/use-dagster-with-teradata.md @@ -0,0 +1,384 @@ +--- +id: use-dagster-with-teradata +sidebar_position: 4.5 +author: Mohan Talla, Daniel Herrera +email: developer.relations@teradata.com +page_last_update: 2026-06-23 +description: Use dagster-teradata with Teradata. +keywords: [dagster, dagster-teradata, data warehouses, compute storage separation, teradata, vantage, cloud data platform, object storage, business intelligence, enterprise analytics, elt] +--- + +import TrialDocsNote from '../_partials/teradata_trial.mdx' + +# dagster-teradata with Teradata + +This guide walks you through integrating Dagster with Teradata to create and manage ETL pipelines. It provides step-by-step instructions for installing and configuring the necessary packages, setting up a Dagster project, and implementing a pipeline that interacts with Teradata. + +## Dagster + +* Dagster is a data orchestrator built for data engineers, with integrated lineage, observability, a declarative programming model and best-in-class testability. +* Data pipelines are automated workflows that ingest raw data, process it through various transformations (such as cleaning and structuring), and produce a final, usable format—much like an assembly line for data. +* Dagster orchestrates this process by defining each stage of the pipeline, ensuring tasks execute in the correct sequence and at scheduled intervals. It provides a structured way to manage dependencies, track execution, and maintain reliable data workflows. +* Dagster orchestrates dbt alongside other technologies. Dagster's asset-oriented approach allows Dagster to understand dbt at the level of individual dbt models. + + +## Prerequisites + +* Access to a Teradata cloud or on-premises instance (Teradata Cloud, Teradata Factory, or Teradata Trial). + + + +* Python **3.9** or higher, Python **3.12** is recommended. +* `uv` package manager for Python environment management. +* A Teradata database where you have CREATE TABLE privileges. You can create one by running: + + ```sql + CREATE DATABASE dagster_pipeline_db AS PERMANENT = 100e6; + ``` + +## Setting Up the Project with `uv` + +We'll use `uv` exclusively to manage dependencies and run commands. No manual venv activation is required. + +## Initialize a Dagster Project + +We'll use `uvx` to scaffold a new Dagster project, which automatically creates a `pyproject.toml` with all dependencies. + +### Create a New Dagster Project + +Run the following command: + +```bash +uvx create-dagster@latest project dagster-quickstart +``` + +When prompted, respond `y` to run `uv sync` which will set up the isolated environment and install all dependencies: + +``` +A `uv` installation was detected. Run `uv sync`? (y/n) [y]: y +``` + +This command will create a new project named `dagster-quickstart` with the following directory structure: + +```bash +cd dagster-quickstart +``` + +```bash +dagster-quickstart +│ pyproject.toml +│ README.md +│ uv.lock +│ .gitignore +│ +├───.venv/ (virtual environment created by uv sync) +│ +├───.dg/ (Dagster CLI configuration) +│ +├───src/ +│ └───dagster_quickstart/ +│ ├── definitions.py +│ ├── defs/ +│ │ └── __init__.py +│ └── __init__.py +│ +└───tests/ + └── __init__.py +``` + +### Configure the `pyproject.toml` with Required Packages + +The generated `pyproject.toml` needs the `dagster-teradata` package to interact with Teradata. Open the `pyproject.toml` file and add `dagster-teradata` to the dependencies section: + +```toml +dependencies = [ + ... + "pandas", + "dagster-teradata", + "python-dotenv", +] +``` + +After modifying the `pyproject.toml`, run `uv sync` to install the new dependencies: + +```bash +uv sync +``` + +This ensures that all required packages, including `dagster-teradata`, are installed in your isolated environment. + +## Create Sample Data + +To simulate an ETL pipeline, create a CSV file with sample data that your pipeline will process. + +**Create the data directory:** First, create a `data` directory inside the `dagster_quickstart` project root: + +```bash +mkdir data +``` + +**Create the CSV File:** Inside the `/data` directory, create a file named `sample_data.csv` with the following content: + +``` +id,name,age,city +1,Alice,28,New York +2,Bob,35,San Francisco +3,Charlie,42,Chicago +4,Diana,31,Los Angeles +``` + +This file represents sample data that will be used as input for your ETL pipeline. + +## Create a Database for the Pipeline + +Before defining assets, create a database where the pipeline can create and drop tables: + +```sql +CREATE DATABASE dagster_pipeline_db AS PERMANENT = 100e6; +``` + +## 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: + +```python +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. + +### Register Assets in `defs/__init__.py` + +Now you need to register these assets so Dagster can discover them. Update the existing `defs/__init__.py` file and add the following: + +```python +from .assets import read_csv_file, read_table, create_table, drop_table, insert_rows + +__all__ = [ + "read_csv_file", + "read_table", + "create_table", + "drop_table", + "insert_rows", +] +``` + +This makes the assets importable from the `defs` module, allowing them to be discovered by Dagster's asset lineage system. + +## Set Up Environment Variables + +Before defining the pipeline, configure the environment variables that the Teradata resource will use to connect to your Teradata instance. Create a `.env` file in the root of your `dagster-quickstart` project with the following content: + +```bash +TERADATA_HOST=your_teradata_host +TERADATA_USER=your_teradata_username +TERADATA_PASSWORD=your_teradata_password +TERADATA_DATABASE=dagster_pipeline_db +``` + +Replace the placeholder values with your actual Teradata connection details: +- `TERADATA_HOST`: The hostname or IP address of your Teradata instance +- `TERADATA_USER`: Your Teradata username +- `TERADATA_PASSWORD`: Your Teradata password +- `TERADATA_DATABASE`: The database name (use `dagster_pipeline_db` if you created it as shown in the prerequisites) + + +The next step is to configure the pipeline by defining the necessary resources and jobs. + +**Edit the definitions.py File:** Modify `src/dagster_quickstart/definitions.py` and define your Dagster pipeline as follows: + +```python +import os +from dotenv import load_dotenv + +from dagster import Definitions +from dagster_teradata import TeradataResource + + +from .defs import read_csv_file, read_table, create_table, drop_table, insert_rows + + +# Load environment variables from .env file +load_dotenv() + +# Configure Teradata resource with connection details from environment variables +td_resource = TeradataResource( + host=os.getenv("TERADATA_HOST"), + user=os.getenv("TERADATA_USER"), + password=os.getenv("TERADATA_PASSWORD"), + database=os.getenv("TERADATA_DATABASE"), +) + +# Define the pipeline and resources +defs = Definitions( + assets=[read_csv_file, read_table, create_table, drop_table, insert_rows], + resources={ + "teradata": td_resource, + } +) +``` + +This code sets up a Dagster project that interacts with Teradata by defining assets and resources: + +1. It imports necessary modules, including Dagster and dagster-teradata. +2. It imports asset functions (read_csv_file, read_table, create_table, drop_table, insert_rows) from the defs module. +3. It configures the TeradataResource with connection details from environment variables. +4. It registers these assets with Dagster using `Definitions`, allowing Dagster to track and execute them. + +## Running the Pipeline + +After setting up the project, you can now run your Dagster pipeline: + +1. **Start the Dagster Dev Server:** In your terminal, navigate to the root directory of your project and run: + + ```bash + uv run dg dev + ``` + + The `uv run` command ensures that `dg dev` runs within the project's isolated environment defined in `pyproject.toml`. No manual venv activation is needed. + + After executing the command, the Dagster logs will be displayed in the terminal. Once you see a message similar to: + + ```bash + 2025-02-04 09:15:46 +0530 - dagster - INFO - Serving Dagster UI on http://127.0.0.1:3000 + ``` + + The Dagster web server is running successfully. + + > **Note:** `dg dev` creates an ephemeral instance by default. To persist your runs and assets across sessions, set the `DAGSTER_HOME` environment variable before running `uv run dg dev`: + > + > **Windows (PowerShell):** + > ```powershell + > $env:DAGSTER_HOME="$env:USERPROFILE\.dagster_home" + > uv run dg dev + > ``` + > + > **macOS/Linux:** + > ```bash + > export DAGSTER_HOME=~/.dagster_home + > uv run dg dev + > ``` + +2. **Access the Dagster UI:** + + Open a web browser and navigate to `http://127.0.0.1:3000`. This will open the Dagster UI where you can manage and monitor your pipelines. + + ![dagster-teradata1.png](../images/dagster/dagster-teradata1.png) + +3. **Run the Pipeline:** + + - In the left navigation of the Dagster UI, click on **Lineage**. + + ![dagster-teradata2.png](../images/dagster/dagster-teradata2.png) + - Click **Materialize all** to execute the pipeline. + +4. **Monitor the Run:** + + The Dagster UI allows you to visualize the pipeline's progress, view logs, and inspect the status of each step. You can switch between different views to see the execution logs and metadata for each asset. + + ![dagster-teradata3.png](../images/dagster/dagster-teradata3.png) + +## TeradataResource Operations + +Below are some of the operations provided by the TeradataResource: + +### 1. Execute a Query (`execute_query`) + +This operation executes a SQL query within Teradata. + +**Args:** +- `sql` (str) – The query to be executed. +- `fetch_results` (bool, optional) – If True, fetch the query results. Defaults to False. +- `single_result_row` (bool, optional) – If True, return only the first row of the result set. Effective only if `fetch_results` is True. Defaults to False. + +### 2. Execute Multiple Queries (`execute_queries`) + +This operation executes a series of SQL queries within Teradata. + +**Args:** +- `sql_queries` (Sequence[str]) – List of queries to be executed in series. +- `fetch_results` (bool, optional) – If True, fetch the query results. Defaults to False. +- `single_result_row` (bool, optional) – If True, return only the first row of the result set. Effective only if `fetch_results` is True. Defaults to False. + +### 3. Drop a Database (`drop_database`) + +This operation drops one or more databases from Teradata. + +**Args:** +- `databases` (Union[str, Sequence[str]]) – Database name or list of database names to drop. + +### 4. Drop a Table (`drop_table`) + +This operation drops one or more tables from Teradata. + +**Args:** +- `tables` (Union[str, Sequence[str]]) – Table name or list of table names to drop. + + +## Summary +This guide provides a step-by-step approach to integrating Dagster with Teradata for building ETL pipelines. + +## Further reading +* https://docs.dagster.io/ +* https://docs.dagster.io/getting-started/quickstart +* https://docs.dagster.io/getting-started/installation +* https://docs.dagster.io/etl-pipeline-tutorial/ + + + +