Apache Airflow MySQL Provider

Warnings

• breaking The `MySqlOperator` was removed from the provider package. Users should migrate to `SQLExecuteQueryOperator` from `airflow.providers.common.sql.operators.sql` for executing SQL queries. This change was introduced in provider versions 6.3.2 and later.
  Fix: Replace `MySqlOperator` imports and usages with `SQLExecuteQueryOperator`. The `conn_id` and `sql` parameters remain the same, and schema can be passed via `hook_params={'schema': '<database>'}`.
• breaking Older versions of the MySQL provider (e.g., 2.x and above) require Apache Airflow version 2.1.0+ due to the removal of the `apply_default` decorator. Provider version 3.0.0+ specifically requires Airflow 2.2.0+. Installing an incompatible provider version may lead to automatic Airflow upgrades and require manual database migration.
  Fix: Ensure your Airflow installation meets the minimum version requirement for the MySQL provider you intend to install. Upgrade Airflow if necessary before installing the provider.
• gotcha Attempting to import MySQL-related components like `MySqlOperator` or `MySqlHook` without explicitly installing `apache-airflow-providers-mysql` will result in a `ModuleNotFoundError`.
  Fix: Always install the provider package using `pip install apache-airflow-providers-mysql` in your Airflow environment.
• gotcha Installation of the `mysqlclient` dependency can fail if required system-level development packages (e.g., `libmysqlclient-dev` on Debian/Ubuntu) are not present, leading to build errors.
  Fix: Before installing the Python package, install the necessary system-level development headers for MySQL. For Debian/Ubuntu, use `sudo apt-get install default-libmysqlclient-dev`.
• gotcha Users have reported `AttributeError: property 'connection' of 'MySqlHook' object has no setter` when `apache-airflow-providers-mysql` has no upper version constraint on `apache-airflow-providers-common-sql`, leading to incompatible versions being installed.
  Fix: Ensure that `apache-airflow-providers-common-sql` is constrained to a compatible version or upgrade `apache-airflow-providers-mysql` to its latest version, which usually resolves such conflicts.
• gotcha Long-running MySQL connections via Airflow can experience 'Server has gone away' errors, typically due to the MySQL server restarting or connection timeouts.
  Fix: For persistent connections, consider enabling connection pooling with recycling. Adjust MySQL server's `wait_timeout` and `interactive_timeout` parameters if applicable.

Install

• pip install apache-airflow-providers-mysql Install the MySQL Provider
• sudo apt-get install default-libmysqlclient-dev # For Debian/Ubuntu # Or for older Debian: sudo apt-get install libmysqlclient-dev System dependencies for mysqlclient (if build errors occur)

Imports

• MySqlHook

  from airflow.providers.mysql.hooks.mysql import MySqlHook

• SQLExecuteQueryOperator

  from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

  `MySqlOperator` was removed in provider versions 6.3.2+; use `SQLExecuteQueryOperator` from the `common.sql` provider instead.
• MySqlToS3Operator

  from airflow.providers.mysql.operators.mysql import MySqlToS3Operator

Quickstart

This quickstart demonstrates how to use `SQLExecuteQueryOperator` to interact with a MySQL database. It creates a table, inserts data, and then selects it. Ensure you have a MySQL connection named `mysql_default` configured in your Airflow UI.

from __future__ import annotations

import pendulum

from airflow.models.dag import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

# Ensure you have a MySQL connection named 'mysql_default' configured in Airflow UI
# Host: localhost, Schema: airflow_db, User: airflow, Pass: airflow

with DAG(
    dag_id="mysql_quickstart_dag",
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    catchup=False,
    schedule=None,
    tags=["mysql", "example"],
) as dag:
    create_table = SQLExecuteQueryOperator(
        task_id="create_test_table",
        conn_id="mysql_default",
        sql="""
            CREATE TABLE IF NOT EXISTS test_table (
                id INT AUTO_INCREMENT PRIMARY KEY,
                name VARCHAR(255)
            );
        """,
    )

    insert_data = SQLExecuteQueryOperator(
        task_id="insert_test_data",
        conn_id="mysql_default",
        sql="INSERT INTO test_table (name) VALUES ('Airflow User 1'), ('Airflow User 2');",
    )

    select_data = SQLExecuteQueryOperator(
        task_id="select_test_data",
        conn_id="mysql_default",
        sql="SELECT * FROM test_table;",
    )

    create_table >> insert_data >> select_data