Skip to main content

Use Measured Cloud with Apache Airflow

Goal​

You'll run Measured Cloud validations from your Airflow DAGs so a failed validation blocks downstream tasks the same way any other Airflow task failure would. Instead of writing a SimpleHttpOperator + custom poller pair, you install the official provider package and use first-class operators and sensors.

The provider package source lives at airflow/ in this repo and ships under the name dq-cloud-airflow (FEATURE-26). Publishing to PyPI is on the roadmap; until then, install directly from the repo (see below).

Coming from checkpoints? Checkpoints were retired (see Checkpoints (retired)). DQCloudCheckpointOperator, DQCloudCheckpointResultSensor and DQCloudHook.run_checkpoint still work for one release as deprecated shims: they run the validation the checkpoint maps to and emit a DeprecationWarning. Switch to DQCloudValidationOperator with that validation_id (shown on the checkpoint in GET .../checkpoints/{id}). The shims are removed in the next major release.

Prereqs​

  • Apache Airflow >=2.7,<3 running somewhere — Astronomer, MWAA, self-hosted, doesn't matter.
  • Python 3.9 or newer in the Airflow worker environment.
  • A Measured Cloud organisation + workspace + at least one validation to run. Every data asset with rules has a default validation, so to check one table, run its default validation.
  • Workspace editor role or higher in Measured Cloud (so your personal API key can trigger runs).

Steps​

  1. Install the provider in the Airflow worker environment.

    While the package is not yet on PyPI, install it straight from the repo:

    pip install "git+https://github.com/your-org/dq-cloud.git#subdirectory=airflow"

    For a stable install, pin to a tag or commit SHA in the URL.

    Restart the Airflow webserver and scheduler after installing — they pick up new providers at boot.

  2. Generate a personal API key in Measured Cloud.

    In the UI, open Settings → API keys → New API key (see Create an API token). Copy the dqk_… value somewhere safe — you only see it once.

  3. Create an Airflow connection.

    In the Airflow UI: Admin → Connections → +. Pick Measured Cloud as the connection type. Fill in:

    • Measured Cloud API base URL — https://dq.example.com (the host where your Measured Cloud lives, no trailing slash).
    • Personal API key (dqk_…) — paste the value from step 2.

    Save. The default connection ID is dq_cloud_default; if you create the connection under a different ID, pass it as dq_cloud_conn_id in the operator.

  4. Add the operator to a DAG.

    from datetime import datetime

    from airflow import DAG
    from dq_cloud_airflow.operators import DQCloudValidationOperator

    ORG = "00000000-0000-0000-0000-000000000001"
    WORKSPACE = "00000000-0000-0000-0000-000000000002"
    VALIDATION_ORDERS = "00000000-0000-0000-0000-0000000000aa"

    with DAG(
        "nightly_warehouse_validation",
        start_date=datetime(2026, 1, 1),
        schedule="0 2 * * *",
        catchup=False,
    ) as dag:
        validate_orders = DQCloudValidationOperator(
            task_id="validate_orders",
            org_id=ORG,
            workspace_id=WORKSPACE,
            validation_id=VALIDATION_ORDERS,
        )

    The operator starts an execution run of the validation (POST /api/v2/.../validations/{id}/run), polls the run every 10 seconds until it finishes, and raises AirflowException when it ends failed or cancelled (pass fail_on_failure=False to only record it). The finished execution run is the task's return value, so a downstream task can read its status and health score with ti.xcom_pull(task_ids="validate_orders"); the operator also pushes execution_run_id, parent_result_id and validation_id as separate XCom keys. ValidationOperator is a short alias of the same class. Optional arguments: rule_ids (run only some of the validation's rules), batch_definition_id and partition_value (run one partition).

  5. Tune polling for long-running validations.

    The default operator holds an Airflow worker slot while it polls. For validations that take more than a few minutes, start the run from one task with wait_for_completion=False (the task returns the execution_run_id) and watch it with a reschedule-mode sensor in another:

    from dq_cloud_airflow.operators import DQCloudValidationOperator
    from dq_cloud_airflow.sensors import DQCloudExecutionRunSensor

    start_orders = DQCloudValidationOperator(
        task_id="start_orders",
        org_id=ORG,
        workspace_id=WORKSPACE,
        validation_id=VALIDATION_ORDERS,
        wait_for_completion=False,
    )
    wait_for_orders = DQCloudExecutionRunSensor(
        task_id="wait_for_orders",
        org_id=ORG,
        workspace_id=WORKSPACE,
        execution_run_id=start_orders.output,
        mode="reschedule",
        poke_interval=60,
    )
    start_orders >> wait_for_orders

    ExecutionRunSensor is a short alias of DQCloudExecutionRunSensor.

Verify​

  • The DAG run shows the task starting, the Logs tab streams DQ Cloud execution run … status=running lines every 10 seconds, and the task turns green once the run reaches succeeded.
  • In Measured Cloud, a new run appears on the validation's Execution runs card (trigger api), and its results on the Results page.
  • Pull the XCom: ti.xcom_pull(task_ids="validate_orders") returns the finished execution run (status, health score, asset counts).

Troubleshooting​

  • AirflowException: 401 — the dqk_… key was rejected. Confirm it hasn't been revoked in Settings → API keys, and that the workspace it's scoped to matches the workspace_id in the operator.
  • AirflowException: missing 'host' — the Airflow connection was saved without a base URL. Re-open it, paste the URL, save.
  • Task hangs at status=pending — Measured Cloud accepted the run but no worker has picked it up. Check the Measured Cloud worker logs. If the validation includes an agent-mode datasource, confirm the on-prem agent is connected: the run stays running until the agent reports.
  • Task fails with a 409 — the validation has nothing to run (every rule disabled, or its assets gone). With the deprecated checkpoint operator, a 409 also means the checkpoint maps to no validation.
  • Task fails with did not reach a terminal status within Ns — bump the operator's timeout argument, or move to the sensor-based pattern in step 5 so polling doesn't hold a worker slot.

Out of scope​

  • A @task-style TaskFlow decorator wrapper is a follow-up — see the FEATURE-26 notes.
  • A dbt operator lives in FEATURE-27, tracked separately.