Skip to content
Contact us Go to Customer Portal
  • There are no suggestions because the search field is empty.

Oracle - Load data - Apache Airflow Load Control

Applies to
Platform: Oracle

Load the deployed Oracle solution with Apache Airflow by running the DAG that the generator writes. The DAG has one task per loader, runs them in dependency order and calls each loader through the Python driver oracledb. Afterwards each task log shows the row counts of its loader.

Prerequisites

Run the DAG

  1. Copy LoadControl\<project>_dag.py from the replaced artifacts into the folder dags of Airflow. <project> is the project name in lower case with underscores, for example test_datavault_oracle.
  2. Create one connection per layer. The connection ID is the schema_name value of the layer from the placeholder values. Enter the layer user as login and the service name as schema: 
    docker compose exec airflow-scheduler airflow connections add `
  <schema name> --conn-type generic `
  --conn-host host.docker.internal --conn-port 1521 `
  --conn-login <user name> --conn-password <password> `
  --conn-schema <service name>
    The command prints Successfully added `conn_id`=<schema name>. Repeat it for each layer.
  3. Unpause the DAG: docker compose exec airflow-scheduler airflow dags unpause <dag id>. The DAG ID is the file name without _dag.py.
  4. Trigger it: docker compose exec airflow-scheduler airflow dags trigger <dag id>. The run starts in the state queued.

Each Airflow task passes its own load timestamp, taken from the Airflow worker clock in whole seconds. The native load control instead uses one load timestamp for all loaders of an execution, see Oracle - Load data - Native Load Control.

Check the result

docker compose exec airflow-scheduler airflow dags list-runs -d <dag id> shows the run with the state success when all tasks succeeded. The log of each task lists the loader and its row counts, for example Inserted: 37. When a password equals the schema name, Airflow masks the schema name in the log as ***.

Example

The DAG test_datavault_oracle has 105 loader tasks between the tasks begin and end. Each task calls one loader with the connection of its layer:

op_kwargs = {"procedure_name":
    "DWH_STG.STG_ST_CUSTOMER_LOADER",
    "connection_id": "DWH_STG"}

Here DWH_STG is the example Stage schema. With Airflow 2.9.1 in Docker and a local database, the run took about three and a half minutes and ended with success.

Failure symptoms and recovery

The DAG shows an import error for oracledb

Cause: The Airflow image does not contain the Python package oracledb.
Fix: Add the line oracledb to requirements.txt and rebuild the environment as described in Apache Airflow - Load control environment.

A task fails because a connection is missing

Cause: No Airflow connection has the schema name of the layer as its ID.
Fix: Create the connection as in step 2 and clear the failed task.

A task fails with a network error

Message: DPY-6005
Cause: Airflow could not reach the database host or port.
Fix: Check host and port of the connection. Then rerun the task and its downstream tasks: docker compose exec airflow-scheduler airflow tasks clear <dag id> -t "^<task id>$" -d -y.

Release differences

  • Generator 2.2: the DAG calls the loaders through oracledb. Up to Generator 2.0.1 it imported the Microsoft SQL Server provider. See Generators Release - 2.2.0.

Related articles