How to integrate MariaDB with Airflow in PostgreSQL

What problem does MariaDB integration with Airflow solve?

It schedules, orchestrates, and monitors SQL jobs against MariaDB, replacing ad-hoc scripts with version-controlled, retry-aware DAGs.

How do I add a MariaDB connection in Airflow?

Use the Airflow UI, CLI, or environment variables. Provide host, port (3306), user, password, and schema. Choose the “mysql” conn_type—MariaDB is wire-compatible.

Airflow CLI example

airflow connections add mariadb_ecom \
--conn-uri 'mariadb://analytics:SECRETPW@db.prod.local:3306/ecommerce'

What syntax does the MySqlOperator support?

MySqlOperator(sql="…", mysql_conn_id="mariadb_ecom", parameters=None, autocommit=False, database=None, **kwargs) executes SQL strings or files. Use Jinja in sql for templating.

Templated example

add_daily_orders = MySqlOperator(
task_id="agg_daily_orders",
sql="sql/agg_daily_orders.sql",
mysql_conn_id="mariadb_ecom",
parameters={"ds": '{{ ds }}'},
)

How do I read data inside a Python task?

MySqlHook offers get_records(), get_pandas_df(), and run(). It returns lists or DataFrames you can process further.

PythonOperator example

def stale_customers(**ctx):
hook = MySqlHook(mysql_conn_id="mariadb_ecom")
rows = hook.get_records("""
SELECT id, email
FROM Customers
WHERE created_at < DATE_SUB(CURDATE(), INTERVAL 1 YEAR)
""")
# send email or write to log


When should I enable autocommit?

Enable autocommit=True for DDL or large ETL inserts to avoid implicit transaction overhead. Keep it False when you need rollback safety.

Best practices for production DAGs

• Store SQL in version-controlled files, not inline strings.
• Template dates with Airflow macros ({{ ds }}, {{ ds_nodash }}).
• Use dedicated service accounts with least-privilege grants.
• Fail fast—set retries and retry_delay.

Common mistake: missing schema in connection URI

Omitting the database name forces you to fully qualify every table. Add /ecommerce in the URI to default the schema.

Common mistake: forgetting autocommit on multi-statement SQL

Running bulk INSERT … SELECT without autocommit=True can fill transaction logs and lock tables. Enable autocommit or split the job.

How do I test MariaDB tasks locally?

Launch a Docker MariaDB container, set an Airflow .env with the same connection URI, and trigger the DAG with airflow dags test.

Example DAG overview

with DAG(
dag_id="daily_ecommerce_metrics",
schedule_interval="0 2 * * *",
start_date=datetime(2023, 1, 1),
catchup=False,
) as dag:

agg_orders = MySqlOperator(
task_id="agg_daily_orders",
sql="sql/agg_daily_orders.sql",
mysql_conn_id="mariadb_ecom",
autocommit=True,
)

mark_stale = PythonOperator(
task_id="flag_stale_customers",
python_callable=stale_customers,
)

agg_orders >> mark_stale

Want to write SQL faster?

Galaxy’s AI copilot completes SQL, optimizes queries, and shares them across your team—perfect alongside Airflow.