Disclaimer

Your use of this download is governed by Stonebranch’s Terms of Use, which are available at Stonebranch Integration Hub - Terms of Use.

Overview

Apache Airflow is an open source platform created to programmatically author, schedule, and monitor workflows.

This Universal Extension provides the capability to integrate with Apache Airflow and use it as part of your end-to-end Universal Controller workflow, allowing high-level visibility and orchestration of data-oriented jobs or pipelines.

Software Requirements

This integration requires a Universal Agent and a Python runtime to execute the Universal Task.

Software Requirements for Universal Template and Universal Task

Requires Python 3.7.0 or higher. Tested with the Universal Agent bundled Python distribution.

Software Requirements for Universal Agent

Both Windows and Linux agents are supported:

Universal Agent for Windows x64 Version 7.0.0.0 and later with python options installed.
Universal Agent for Linux Version 7.0.0.0 and later with python options installed.

Software Requirements for Universal Controller

Universal Controller Version 7.0.0.0 and later. This Universal Task requires that Universal Controller property Web Service Default Response Content (uc.web_service.response.content.default) is set to JSON.

Supported Apache Airflow Versions

This integration is tested on Apache airflow V2.2.3.

Key Features

This extension provides support for:

Triggering a new DAG run.
Information retrieval of a specific DAG run.
Information retrieval for a task that is part of a specific DAG run.
Basic authentication (username/password) and SSL protocol.
Using a proxy between Universal Controller and Apache Airflow server.

Import the Universal Template

To use this downloadable Universal Template, you first must perform the following steps:

This Universal Task requires the Resolvable Credentials feature. Check that the Resolvable Credentials Permitted system property has been set to true.
Download the provided ZIP file.
In the Universal Controller UI, select Administration >Configuration > Universal Templates to display the current list of Universal Templates.
Click Import Template.
Select the template ZIP file and Import.

When the template has been imported successfully, the Universal Template will appear on the list. Refresh your Navigation Tree to see these tasks in the Automation Center Menu.

Configure Universal Task

For the new Universal Task type, create a new task and enter the task-specific details that were created in the Universal Template.

Input Fields

The input fields for this Universal Extension are described in the following table.

Field	Input type	Default value	Type	Description
Airflow Base URL	Required	-	Text	The Base URL of the Airflow server.
Airflow Credentials	Required	-	Credentials	The Apache Airflow account credentials. They are comprised of: Runtime username Runtime password
Action	Required	Trigger DAG Run	Choice	The action performed upon the task execution. Valid values are: Trigger Dag Run Read DAG Run Information Read Task Instance Information
DAG Id	Required	-	Dynamic Choice	Dynamic Choice field populated by getting a list of active DAG’s from the server.
DAG Run Id	Optional	-	Text	Id of a specific DAG Run. Required for Action "Read DAG Run Information"/"Read Task Instance Information".
Task Id	Optional	-	Text	Dynamic Choice field populated by getting a list of Task Ids for a specific DAG ID. Required for Action "Read Task Instance Information"
Use Proxy	Required	False	Boolean	Flag to allow Proxy configuration. Required when connection to Apache Airflow is through Proxy.
Proxy Servers	Optional	-	Text	Proxy server and port. Valid format: http://proxyserver:port or https://proxyserver:port. Required when Use Proxy is True.
Use SSL	Required	False	Boolean	Specifies if SSL protocol should be used for the communication with the foreign API.
SSL Hostname Check	Optional	True	Boolean	Determines if the host name of the certificate should be verified against the hostname in the URL. Required when Use SSL is checked.
SSL Certificate Path	Optional	-	Text	Path and file name of the trusted certificate or CA bundle to use in certificate verification. The file must be in PEM format.

Task Examples

Trigger a new DAG Run

Example of Universal Task for triggering a new DAG Run. In this specific example SSL protocol, as well as a proxy, is used.

Extension Output Sample

{
    "exit_code": 0,
    "status_description": "SUCCESS: Successful Task Execution",
    "changed": true,
    "invocation": {
        "extension": "ue-airflow",
        "version": "1.0.0",
        "fields": {
            "credentials_user": "****",
            "credentials_password": "****",
            "base_url": "https://foo_airflow_base_url.com:8443/api/v1",
            "use_ssl": true,
            "ssl_verify": true,
            "trusted_certificate_file": "C:\\Users\\****\\temp\\some_certificate.cer",
            "ssl_hostname_check": false,
            "private_key_certificate": null,
            "public_key_certificate": null,
            "use_proxy": true,
            "proxies": "http://foo.proxy.com:8080",
            "action": "trigger_dag_run",
            "dag_id": "example_bash_operator",
            "dag_run_id": null,
            "task_id": null
        }
    },
    "result": {
        "conf": {},
        "dag_id": "example_bash_operator",
        "dag_run_id": "manual__2022-02-09T13:09:48.136996+00:00",
        "end_date": null,
        "execution_date": "2022-02-09T13:09:48.136996+00:00",
        "external_trigger": true,
        "logical_date": "2022-02-09T13:09:48.136996+00:00",
        "start_date": null,
        "state": "queued"
    }
}

Read DAG Run information

Example of Universal Task for getting DAG Run information

Extension Output Sample

{
    "exit_code": 0,
    "status_description": "SUCCESS: Successful Task Execution",
    "changed": false,
    "invocation": {
        "extension": "ue-airflow",
        "version": "1.0.0",
        "fields": {
            "credentials_user": "****",
            "credentials_password": "****",
            "base_url": "https://foo_airflow_base_url.com:8443/api/v1",
            "use_ssl": true,
            "ssl_verify": false,
            "trusted_certificate_file": "/home/****/temp/airflow_cert.cer",
            "ssl_hostname_check": false,
            "private_key_certificate": null,
            "public_key_certificate": null,
            "use_proxy": false,
            "proxies": null,
            "action": "get_dag_run",
            "dag_id": "example_bash_operator",
            "dag_run_id": "manual__2022-02-08T09:17:54.600827+00:00",
            "task_id": null
        }
    },
    "result": {
        "conf": {},
        "dag_id": "example_bash_operator",
        "dag_run_id": "manual__2022-02-08T09:17:54.600827+00:00",
        "end_date": null,
        "execution_date": "2022-02-08T09:17:54.600827+00:00",
        "external_trigger": true,
        "logical_date": "2022-02-08T09:17:54.600827+00:00",
        "start_date": null,
        "state": "queued"
    }
}

Reading Airflow Task Instance information

Example of Universal Task for getting information on Airflow Task instance.

Extension Output Sample

{
    "exit_code": 0,
    "status_description": "SUCCESS: Successful Task Execution",
    "changed": false,
    "invocation": {
        "extension": "ue-airflow",
        "version": "1.0.0",
        "fields": {
            "credentials_user": "****",
            "credentials_password": "****",
            "base_url": "https://foo_airflow_base_url.com:8443/api/v1",
            "use_ssl": true,
            "ssl_verify": true,
            "trusted_certificate_file": "/home/****/temp/some_certificate.cer",
            "ssl_hostname_check": false,
            "private_key_certificate": null,
            "public_key_certificate": null,
            "use_proxy": false,
            "proxies": null,
            "action": "get_task_instance",
            "dag_id": "example_bash_operator",
            "dag_run_id": "manual__2022-02-08T12:45:08.835386+00:00",
            "task_id": "run_this_last"
        }
    },
    "result": {
        "dag_id": "example_bash_operator",
        "duration": null,
        "end_date": null,
        "execution_date": "2022-02-08T12:45:08.835386+00:00",
        "executor_config": "{}",
        "hostname": "",
        "max_tries": 0,
        "operator": "DummyOperator",
        "pid": null,
        "pool": "default_pool",
        "pool_slots": 1,
        "priority_weight": 1,
        "queue": "default",
        "queued_when": null,
        "sla_miss": null,
        "start_date": null,
        "state": null,
        "task_id": "run_this_last",
        "try_number": 0,
        "unixname": "****"
    }
}

Task Output

Exit Codes

The exit codes for this Universal Extension are described in the following table.

Exit Code	Status Classification Code	Status Classification Description	Status Description
0	SUCCESS	Successful Execution	SUCCESS: Successful Task Execution
1	FAIL	Failed Execution	FAIL: <Error Description>
2	AUTHENTICATION_ERROR	Authentication Error	AUTHENTICATION_ERROR: <Error Description>
3	CONNECTION_ERROR	Connection Error	CONNECTION_ERROR: <Error Description>
20	DATA_VALIDATION_ERROR	Input fields Validation Error	DATA_VALIDATION_ERROR: <Error Description>
21	REQUEST_FAILURE	HTTP request error	REQUEST_FAILED: <Error Description>

Extension Output

The Extension Output samples are available on the task examples above.

Document References

This document references the following documents:

Name	Location	Description
Universal Templates	https://docs.stonebranch.com/confluence/display/UC72x/Universal+Templates	User documentation for creating Universal Templates in the Universal Controller user interface.
Universal Tasks	https://docs.stonebranch.com/confluence/display/UC72x/Universal+Tasks	User documentation for creating Universal Tasks in the Universal Controller user interface.
Apache Airflow Documentation	Apache Airflow Documentation	User documentation for Apache Airflow.
Apache Airflow API Documentation	Airflow REST API	User Documentation for Airflow REST API.