Apache Airflow Community | Package: apache-airflow-providers-ibm-db2 v0.1.0 (Incubating)
|
ANNOUNCEMENT: We are excited to announce the addition of the IBM Db2 provider to the Apache Airflow ecosystem as an incubating community provider! This provider brings first-class support for IBM Db2 into Airflow workflows, making it straightforward to build, schedule, and monitor data pipelines.
|
Why an IBM Db2 Provider?
IBM Db2 is widely deployed in enterprise environments — banking, insurance, manufacturing, and government — often as the system of record for mission-critical data. Until now, Airflow users who needed to orchestrate workflows against Db2 had to rely on generic JDBC or ODBC hooks, losing the ability to use Airflow's built-in connection management UI, SSL configuration, and schema introspection features that are available for other first-class databases.
This provider fills that gap with a purpose-built Db2Hook and a dedicated SQLAlchemy dialect, giving Db2 users the polished experience.
What's Included in v0.1.0
|
Component
|
Class / ID
|
Description
|
|
Hook
|
airflow.providers.ibm.db2.hooks.db2.Db2Hook
|
Full-featured DbApiHook for IBM Db2 with SSL, autocommit, and executemany support.
|
|
Connection Type
|
ibmdb2
|
Custom Airflow UI fields for host, port, database, credentials, SSL, schema, and auth type.
|
|
SQLAlchemy Dialect
|
airflow.providers.ibm.db2.dialects.db2.Db2Dialect
|
Db2-specific dialect for schema introspection, INSERT/UPSERT generation, and column metadata.
|
Connection UI Fields
· Host: Db2 server hostname or IP (Default: localhost).
· Port: Standard Db2 port (Default: 50000).
· Database: Target database name.
· Username / Password: Db2 credentials.
· Security Protocol: Leave empty or set SSL for encryption.
· SSL Server Certificate: Path to .crt file (SSL only).
· Authentication Type: Leave empty for default, or set SERVER · KERBEROS · GSSPLUGIN.
Installation & Dependencies
Note 1: The PyPI package apache-airflow-providers-ibm-db2 will be available on PyPI once the provider graduates to lifecycle: production as per the Apache Airflow provider lifecycle policy.
Note 2: If you have an existing airflow installation, ensure your DB is up to date using the command
Skip the above migration command if you are installing Airflow for the first time.
During the incubation period (tentative 6-8 months), install directly from source:
Please note: This may take some time to clone the repository
|
pip install -v "git+https://github.com/apache/airflow.git@main#subdirectory=providers/ibm/db2"
|
To install from a specific commit:
|
pip install "git+https://github.com/apache/airflow.git@<commit-sha>#subdirectory=providers/ibm/db2"
|
Once the provider graduates to lifecycle: production (after incubation), it will be published to PyPI and can be installed with:
|
pip install apache-airflow-providers-ibm-db2
|
All dependencies are installed automatically. The table below lists minimum required versions and licenses for reference:
· ibm-db ≥ 3.2.0
· ibm-db-sa ≥ 0.4.0
· apache-airflow-providers-common-sql ≥ 1.32.0
· methodtools ≥ 0.4.7 (MIT)
· apache-airflow ≥ 2.11.0
· apache-airflow-providers-common-compat ≥ 1.12.0
After installing, verify the installation using the command
|
airflow providers list | grep ibm
|
Quickstart Guide & DAG Execution Flow
Step 1 — Configure Connection
Configure connection via environment variable or Airflow UI (Admin -> Connections):
|
export AIRFLOW_CONN_DB2_DEFAULT='db2://user:password@host:50000/database
|
Step 2 — Run SQL with SQLExecuteQueryOperator
|
from airflow.sdk import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from datetime import datetime
with DAG("db2_pipeline", start_date=datetime(2024, 1, 1), schedule=None) as dag:
create_table = SQLExecuteQueryOperator(
task_id="create_sample_table",
conn_id="db2_default",
sql="""
CREATE TABLE employees (
employee_id INTEGER NOT NULL PRIMARY KEY,
first_name VARCHAR(50),
department VARCHAR(50),
salary DECIMAL(10, 2),
hire_date DATE
)
""",
autocommit=True,
)
insert_data = SQLExecuteQueryOperator(
task_id="insert_data",
conn_id="db2_default",
sql="""
INSERT INTO employees VALUES
(1, 'Alice', 'Engineering', 85000.00, '2023-01-15'),
(2, 'Bob', 'Marketing', 70000.00, '2023-03-10')
""",
autocommit=True,
)
create_table >> insert_data
|
Step 3 — Programmatic Access with Db2Hook
|
from airflow.sdk import DAG, task
from airflow.providers.ibm.db2.hooks.db2 import Db2Hook
from datetime import datetime
with DAG("db2_hook_example", start_date=datetime(2024, 1, 1), schedule=None) as dag:
@task
def query_employees() -> int:
hook = Db2Hook(conn_id="db2_default")
records = hook.get_records(
"SELECT employee_id, first_name, salary "
"FROM employees WHERE department = ?",
parameters=("Engineering",),
)
for row in records:
print(f" {row[0]}: {row[1]} — ${row[2]:,.2f}")
return len(records)
query_employees()
|