Overview

Human-in-the-loop (HITL) is a demonstration of how to create and run an Airflow pipeline in Data Analytics and Visualizations Engine (DAVE) that includes a task executed in an External Service with user input.

To demonstrate this process, we create a data cleaning pipeline in DAVE and integrate it with an External Service via Python Airflow Sensors to include human interaction.

More specifically, the scenario consists of the following steps:

  1. Upload a dataset (housing.csv) into DAVE platform file storage

  2. Create a new workflow pipeline consisting of 3 tasks:

    1. Get File Data: Import dataset from file storage

    2. Limit Rows: Reduce dataset to 20 rows

    3. Sensor: Pause pipeline while waiting for user input via External Service

  3. Run the workflow pipeline

  4. When execution reaches the Sensor step, navigate to the External Service

  5. Perform data cleaning (delete rows) via External Service frontend

  6. Mark HITL task as completed

  7. View results

In the following sections, we provide configuration and implementation details for this tutorial.

For a quick start, refer to Requirements and Tutorial.

Creating the Sensor Airflow Step

To connect a pipeline with an External Service that requires user input during execution, the pipeline must be paused until input is received.

This is achieved using a sensor-based mechanism that:

  1. Creates and stores a configuration file for the external task

  2. Periodically monitors task status

  3. Sleeps while the external task is pending

  4. Completes when the external task is finished

Python Sensor

Sensors are a special type of Operator designed to wait for an event.

They can wait for:

  • Time-based conditions

  • File availability

  • External events

In this tutorial, Python Sensors are used to monitor the external task status every 5 seconds.

While the task is pending, the sensor sleeps (reschedule mode). Once the task is completed, the sensor succeeds and the pipeline continues.

Sensor Configuration File

Example configuration file for the Python Sensor:

{
    op_name: "sensor_external",
    ...,
    airflow_op_type: "airflow.sensors.python.PythonSensor",
    external: true,
    ...,
    op_params: {
        ...
        poke_interval: 5,
        timeout: 10600,
        mode: "reschedule",
        ...
    }
}

The full file is available at: https://gitlab.com/aidapt/poc/Airflow-with-external-service/-/blob/dae-integration/documentation/configuration_files/sensor_external.json

Python Sensor Script

Example sensor implementation:

def main(**kwargs):

  op_name = create_collection_name(kwargs)
  db_name = get_user(kwargs.get("dag"))

  df = load_pandas_df(kwargs, "configuration_database", "configuration_storage_collection")

  if df is None or len(df) == 0 or len(df[df["data_collection_name"] == op_name]) == 0:

    data_df = load_pandas_df(kwargs)
    save_pandas_df(data_df, kwargs, col=op_name)

    configuration_json = {
      "pipeline_name": kwargs.get("task_instance").dag_id,
      "database_name": db_name,
      "data_collection_name": op_name,
      "status": "pending",
      "task": "",
      "started_at": datetime.today().replace(microsecond=0),
      "ended_at": None
    }

    df = pd.DataFrame([configuration_json])
    save_pandas_df(df, kwargs, "configuration_database", "configuration_storage_collection")

  else:
    df = df[df["data_collection_name"] == op_name]

  if df['status'].iloc[0] != "pending":
    condition_met = True
    operator_return_value = {"dag_id": db_name}
  else:
    condition_met = False
    operator_return_value = None

  return PokeReturnValue(
    is_done=condition_met,
    xcom_value=operator_return_value
  )

Full script: https://gitlab.com/aidapt/poc/Airflow-with-external-service/-/blob/dae-integration/documentation/scripts/sensor_external.py

Integrating The External Service

General Guidelines

To integrate an External Service in DAVE, a middleware database is required between:

  • DAVE pipeline storage

  • External Service

The sensor step:

  • Writes configuration and pipeline state to the database

  • External service reads this data

  • User performs actions via frontend

  • External service updates status

  • Pipeline resumes execution

The External Service consists of:

  1. Backend API (FastAPI)

  2. Frontend UI (React)

Backend

The backend:

  • Connects to configuration storage database

  • Retrieves HITL task configuration

  • Provides dataset access

  • Updates task status

Configuration Schema

class ConfigurationFiles(Document, ConfigurationFileUpdate):

  class Settings:
    name = CONFIG.MONGO_CONFIGURATION_STORAGE_COLLECTION

  pipeline_name: str | None = None
  description: str | None = None
  dataset_name: str | None = None
  data_collection_name: str | None = None
  database_stored: str | None = None
  status: str
  task: str | None = None
  external_service: str | None = None
  started_at: datetime | None = None
  ended_at: datetime | None = None

Get Configuration Endpoint

@router.get("/configuration/document/", status_code=200)
async def get_all_configuration_documents() -> list[ConfigurationFiles]:

  configuration_documents = await ConfigurationFiles.find_all().to_list()
  return configuration_documents

Update Configuration Endpoint

@router.patch("/configuration/document/{configuration_doc_id}", status_code=200)
async def update_configuration_document(
        configuration_document_update: ConfigurationFileUpdate,
        configuration_doc_id: PydanticObjectId) -> ConfigurationFiles:

  configuration_document_to_update = await ConfigurationFiles.get(configuration_doc_id)

  if not configuration_document_to_update:
    return HTTPException(status_code=404, detail="Configuration Document does not exist")

  configuration_document_to_update.status = configuration_document_update.status
  configuration_document_to_update.ended_at = datetime.today().replace(microsecond=0)

  await configuration_document_to_update.save()

  return configuration_document_to_update

Retrieve Dataset Endpoint

@router.get("/storage/{collection_name}/dataset", status_code=200)
async def get_all_storage_datasets(collection_name: str) -> list[Dataset]:

    Dataset.Settings.name = collection_name
    await Database.init_beanie_storage()
    datasets = await Dataset.find_all().to_list()
    return datasets

Delete Dataset Row Endpoint

@router.delete("/storage/{collection_name}/dataset/{dataset_id}", status_code=200)
async def delete_storage_dataset_row(collection_name: str, dataset_id: PydanticObjectId):

    Dataset.Settings.name = collection_name
    await Database.init_beanie_storage()
    dataset = await Dataset.get(dataset_id)

    if dataset is None:
        return HTTPException(status_code=404, detail="Dataset does not exist")

    await dataset.delete()
    return {"message": "Dataset row deleted successfully"}

Frontend

The frontend:

  • Displays incoming HITL tasks

  • Provides task-specific UI actions

  • Allows user to complete tasks

  • Sends updates to backend

Implemented using React JS.

Deployment

Requirements

Hardware

  • Memory: 16 GB RAM

  • CPU: Intel i5-6500 / AMD Ryzen 5 3600 or higher

  • Storage: 15 GB free space

Software

  • DAVE

  • Docker >= 27.4.0

  • Docker Compose >= 2.29.2

Installation

  1. Clone and install DAVE following: https://gitlab.com/Aidapt/data-ai-pipeline-monitoring-services/data-analytics-environment/-/blob/main/docs/resources/initial_setup.md

  2. Navigate to DAVE root directory

  3. Clone external service project:

    git clone -b external-service --single-branch https://gitlab.com/Aidapt/poc/Airflow-with-external-service.git

  4. Navigate to external-service/

  5. Configure environment variables in .env-template

  6. Rename .env-template → .env

  7. Start services:

    docker-compose up -d

Services

Service URL

External Service UI

http://localhost:5173

External Service Backend Docs

http://localhost:8001/docs

Tutorial

Full tutorial: https://gitlab.com/aidapt/poc/Airflow-with-external-service/-/tree/dae-integration

Tutorial Video

Appendix

Terminology

Term Explanation

HITL

Human-in-the-loop

Aidapt component

External or internal application (e.g. data cleaning service)

Step

Task in Airflow pipeline

Airflow Operator

Task template in Airflow

External Service

External application integrated with pipeline

Pipeline Storage

Database storing pipeline data

Configuration Storage

Database storing configuration files

Environment Variables

Variable Example Description

EXTERNAL_FRONTEND_HOST

http://localhost

Frontend host

EXTERNAL_FRONTEND_PORT

5173

Frontend port

REACT_APP_EXTERNAL_BACKEND_HOST

http://localhost

Backend host

REACT_APP_EXTERNAL_BACKEND_PORT

8001

Backend port

MONGO_HOST

mongodb

Mongo host

MONGO_PORT

27017

Mongo port

MONGO_INITDB_ROOT_USERNAME

Aidapt

Mongo username

MONGO_INITDB_ROOT_PASSWORD

Aidapt

Mongo password

MONGO_INITDB_DATABASE

Aidapt

Mongo database

MONGO_CONFIGURATION_STORAGE_COLLECTION

configuration_storage_collection

Config collection

Technologies Used

  • California Housing Dataset

  • Docker

  • Docker Compose

  • React

  • FastAPI

  • Vite

  • Airflow Sensors

  • Python Sensors