Ozone provider

Elena Kostyuchenko

Overview

The Ozone provider enables DAGs to interact with Apache Ozone. It provides hooks, operators, sensors, and transfer operators for performing administrative operations, managing files and directories, and transferring data between Ozone and other storage systems.

The provider can be used for performing the following tasks in Ozone:

  • create and delete volumes and buckets;

  • upload, download, move, copy, and delete files;

  • manage directories;

  • transfer data from HDFS to Ozone;

  • create snapshot-based backups;

  • monitor file and directory availability.

The provider follows the standard Airflow provider architecture, where operators implement DAG tasks, hooks provide low-level communication with external systems, sensors wait until a specified condition becomes true, transfer operators move data between storage systems, and utility modules provide reusable helper functions.

IMPORTANT
This provider does not implement the S3-compatible API and does not use the Ozone S3 Gateway.

Requirements

The Ozone provider requires the following:

  • Airflow 2.10.3 or later;

  • an Ozone connection configured in Airflow;

  • Ozone Client installed on hosts with Airflow workers;

  • network connectivity to Ozone.

To work with the HdfsToOzoneOperator operator, the following additional requirements must be met:

  • HDFS Client installed on hosts with Airflow workers;

  • available Hadoop DistCp;

  • network connectivity to HDFS.

If you already have a working ADH cluster, the required Ozone Client and HDFS Client components can be shared with the ADO cluster by using shared hosts.

Only the client components are required on Airflow hosts. The Ozone Manager, Storage Container Manager, and DataNode components are not required.

Architecture

The Ozone provider enables DAGs to interact with Ozone using the native Ozone command-line tools. It provides hooks, operators, sensors, and transfer operators for managing Ozone storage, monitoring filesystem objects, and transferring data between HDFS and Ozone.

The provider is organized into several Python modules:

  • Hooks

    Hooks implement the communication layer between Airflow and Ozone. They prepare the runtime environment, apply connection settings, and execute native Ozone commands.

    The provider includes the following hook implementations:

    • OzoneAdminHook — performs administrative operations such as creating volumes and buckets, configuring quotas, and managing snapshots.

    • OzoneFsHook — performs filesystem operations including uploading, downloading, copying, moving, listing, and deleting objects.

  • Operators

    Operators implement executable Airflow tasks. Each operator validates the supplied parameters, initializes the appropriate hook, executes the requested operation, and returns the execution result to Airflow.

  • Sensors

    Sensors periodically check the state of Ozone resources until a specified condition becomes true or the configured timeout expires. Typical use cases include waiting for a file produced by another workflow before starting downstream processing.

  • Transfers

    Transfer operators integrate Ozone with external storage systems. The current provider includes operators for copying data from HDFS to Ozone using Hadoop DistCp and for creating Ozone snapshot backups.

These modules work together during DAG execution. Airflow schedules an operator, which initializes the appropriate hook using the configured Airflow connection. The hook prepares the execution environment, invokes the required Ozone CLI command, and returns the execution result to the operator. The operator then reports the task status back to the Airflow scheduler.

Configuration

The provider uses the Ozone Airflow connection type to store the parameters required for communicating with an Ozone cluster.

You can configure the connection in the Airflow UI.

Standard configuration

The provider uses the standard Airflow connection fields shown below. Provider-specific configuration parameters are stored in the Extra field in JSON format.

Field Description

Connection type

The type of connection available in a provider. Must be set to Ozone

Host

Hostname or IP address of Ozone Manager

Port

Port of Ozone Manager

Extra

Provider-specific runtime configuration

Example of a basic connection:

  • Connection Type: ozone

  • Host: ozone-manager.example.com

  • Port: 9862

  • Extra:

{
  "ozone_conf_dir": "/etc/ozone/conf"
}

Extra parameters

The provider reads additional runtime settings from the Extra field.

Parameter Description Default value

ozone_conf_dir

Directory containing the Ozone/HDFS client configuration files (ozone-site.xml and core-site.xml). The provider uses it to set OZONE_CONF_DIR and HADOOP_CONF_DIR

 — 

ozone_security_enabled

Enables Ozone security mode in provider runtime. Set to true for SSL and SSL with Kerberos scenarios

false

ozone_om_https_port

HTTPS port used by Ozone OM in the ADH cluster (for example, 9879 in local setups). Must match actual Ozone SSL config

 — 

hadoop_security_authentication

Authentication mode. Set to kerberos to enable Kerberos authentication

simple

max_content_size_bytes

Maximum size of uploaded or downloaded content

1073741824 (1 GiB)

SSL/TLS

The provider supports encrypted communication with the Ozone cluster. SSL/TLS parameters are specified in the connection Extra field.

The following example configures an SSL connection:

{
  "ozone_security_enabled": "true",
  "ozone_om_https_port": "9879",
  "ozone_ssl_keystore_location": "/etc/ssl/ozone-keystore.jks",
  "ozone_ssl_keystore_password": "secret://vault/ozone/keystore_password",
  "ozone_ssl_keystore_type": "JKS",
  "ozone_ssl_truststore_location": "/etc/ssl/ozone-truststore.jks",
  "ozone_ssl_truststore_password": "secret://vault/ozone/truststore_password",
  "ozone_ssl_truststore_type": "JKS",
  "ozone_conf_dir": "/etc/ozone/conf"
}

The available SSL parameters are listed below.

Parameter Description Default value

ozone_ssl_keystore_location

Path to the SSL keystore

 — 

ozone_ssl_keystore_password

Password for the keystore

 — 

ozone_ssl_keystore_type

Store format (typically JKS)

JKS

ozone_ssl_truststore_location

Path to the SSL truststore

 — 

ozone_ssl_truststore_password

Password for the truststore

 — 

ozone_ssl_truststore_type

Truststore format

JKS

Kerberos

The provider supports Kerberos authentication for secure Ozone deployments.

Configure Kerberos parameters in the connection Extra field:

{
  "hadoop_security_authentication": "kerberos",
  "kerberos_principal": "airflow@EXAMPLE.COM",
  "kerberos_keytab": "/etc/security/keytabs/airflow.keytab",
  "krb5_conf": "/etc/krb5.conf",
  "ozone_conf_dir": "/etc/ozone/conf"
}

The following Kerberos parameters are supported.

Parameter Description Default value

kerberos_principal

Kerberos principal used by Airflow tasks

 — 

kerberos_keytab

Path to the Kerberos keytab

 — 

kerberos_password

Password used instead of a keytab

 — 

krb5_conf

Path to the krb5.conf file

System default

Either kerberos_keytab or kerberos_password must be specified when Kerberos authentication is enabled.

Example configuration for SSL/TLS and Kerberos using keytab
{
  "ozone_security_enabled": "true",
  "ozone_om_https_port": "9879",
  "ozone_ssl_keystore_location": "/opt/airflow/ssl/ozone-keystore.jks",
  "ozone_ssl_keystore_password": "secret://vault/ozone/keystore_password",
  "ozone_ssl_keystore_type": "JKS",
  "ozone_ssl_truststore_location": "/opt/airflow/ssl/ozone-truststore.jks",
  "ozone_ssl_truststore_password": "secret://vault/ozone/truststore_password",
  "ozone_ssl_truststore_type": "JKS",
  "hadoop_security_authentication": "kerberos",
  "kerberos_principal": "testuser@EXAMPLE.COM",
  "kerberos_keytab": "/opt/airflow/keytabs/testuser.keytab",
  "krb5_conf": "/opt/airflow/kerberos-config/krb5.conf",
  "ozone_conf_dir": "/opt/airflow/ozone-conf"
}
Example configuration for SSL/TLS and Kerberos using password
{
  "ozone_security_enabled": "true",
  "ozone_om_https_port": "9879",
  "ozone_ssl_keystore_location": "/opt/airflow/ssl/ozone-keystore.jks",
  "ozone_ssl_keystore_password": "secret://vault/ozone/keystore_password",
  "ozone_ssl_keystore_type": "JKS",
  "ozone_ssl_truststore_location": "/opt/airflow/ssl/ozone-truststore.jks",
  "ozone_ssl_truststore_password": "secret://vault/ozone/truststore_password",
  "ozone_ssl_truststore_type": "JKS",
  "hadoop_security_authentication": "kerberos",
  "kerberos_principal": "testuser@EXAMPLE.COM",
  "kerberos_password": "secret://vault/ozone/kerberos_password",
  "krb5_conf": "/opt/airflow/kerberos-config/krb5.conf",
  "ozone_conf_dir": "/opt/airflow/ozone-conf"
}

HDFS transfer configuration

HdfsToOzoneOperator uses an HDFS connection in addition to the Ozone connection. Security settings required for Hadoop DistCp are read from the connection referenced by hdfs_conn_id.

When Kerberos authentication is enabled for HDFS, configure the HDFS connection with the appropriate principal and either a keytab or password.

Upload and download limits

The provider limits the maximum size of uploaded and downloaded content using the max_content_size_bytes connection parameter.

Individual upload and download operators also provide a max_content_size_bytes parameter that overrides the connection-level setting for a single task.

Hooks

Hooks implement the communication layer of the provider. Since the Ozone provider follows a CLI-based execution model, hooks prepare the runtime environment, configure the execution environment, invoke native CLI commands, and return the execution results to operators and sensors.

The provider includes the following hooks.

Hook Description

OzoneCliHook

Base hook responsible for parsing Airflow connections, preparing the runtime environment, applying SSL/TLS and Kerberos settings, executing native Ozone CLI commands with retries and timeouts, and testing Airflow connections

OzoneFsHook

Extends OzoneCliHook and provides filesystem operations through ozone fs, including creating directories, uploading and downloading files, copying, moving, deleting, listing, and checking the existence of filesystem objects

OzoneAdminHook

Extends OzoneCliHook and provides administrative operations through ozone sh, including volume and bucket management, quota configuration, metadata retrieval, and resource listing

OzoneAdminExtraHook

Extends OzoneAdminHook with advanced administrative operations such as snapshot management, tenant management, bucket replication configuration, bucket links, and Storage Container Manager (SCM) reporting

Operators

Operators implement executable Airflow tasks that perform Ozone operations. Each operator validates its parameters, initializes the appropriate hook, executes the requested CLI command, and returns the execution result to Airflow.

Depending on the operation, operators use either OzoneAdminHook or OzoneFsHook. Data transfer operators additionally invoke Hadoop DistCp utilities where required.

Administrative operators

Administrative operators manage Ozone volumes, buckets, and quotas.

Operator Description

OzoneCreateVolumeOperator

Creates an Ozone volume

OzoneCreateBucketOperator

Creates a bucket within an existing volume

OzoneSetQuotaOperator

Applies a storage quota to a volume or bucket

OzoneDeleteVolumeOperator

Deletes an existing volume

OzoneDeleteBucketOperator

Deletes an existing bucket

Filesystem operators

Filesystem operators execute ozone fs commands to manage directories and objects stored in Ozone.

Operator Description

OzoneCreatePathOperator

Creates a directory in the Ozone filesystem

OzoneUploadContentOperator

Uploads inline text content as a new object

OzoneUploadFileOperator

Uploads a local file to Ozone

OzoneDownloadFileOperator

Downloads an object from Ozone to the local filesystem

OzoneDeleteKeyOperator

Deletes one or more objects

OzoneDeletePathOperator

Deletes a directory

OzoneListOperator

Lists objects within a directory

OzoneMoveOperator

Moves or renames an object

OzoneCopyOperator

Copies an object within Ozone

OzonePathExistsOperator

Checks whether a file or directory exists

Most filesystem operators support configurable behavior when the destination already exists. Depending on the selected policy, the operator can fail, ignore the existing object, or overwrite uploaded files.

Upload and download operators also support configurable size limits through the max_content_size_bytes parameter. If this parameter is not specified explicitly, the provider uses the value configured in the Airflow connection.

Transfer operators

Transfer operators integrate Ozone with external storage systems and backup functionality.

Operator Description

HdfsToOzoneOperator

Transfers data from HDFS to Ozone using Hadoop DistCp

OzoneBackupOperator

Creates an Ozone bucket snapshot for backup purposes

HdfsToOzoneOperator reads HDFS-specific security settings from the connection referenced by hdfs_conn_id and applies them only to the DistCp execution environment.

Sensors

Sensors allow DAGs to wait until a resource becomes available before continuing workflow execution.

The provider currently includes the following sensor.

Sensor Description

OzoneKeySensor

Waits until a specified object or directory becomes available in the Ozone filesystem

OzoneKeySensor uses OzoneFsHook internally to periodically check whether the requested path exists.

The sensor supports templated paths, allowing DAGs to monitor dynamically generated locations. During every polling cycle, the sensor executes a lightweight CLI existence check using the same runtime configuration and security settings as all other provider components.

Unlike the standard Airflow sensor timeout, the timeout parameter of OzoneKeySensor specifies the timeout for a single CLI execution. Overall sensor execution behavior, including polling frequency and total waiting time, is controlled by the standard Airflow sensor parameters such as poke_interval and the inherited BaseSensorOperator settings.

Use the provider in a DAG

To use the Ozone provider, import the required operators or sensors from the provider package and configure an Ozone connection.

The provider includes several example DAGs that demonstrate common administrative and data engineering workflows. These examples can be executed without modifying the DAG source code. Runtime parameters can be overridden from the Airflow Trigger DAG dialog or by passing values in dag_run.conf.

Basic storage workflow

The example_ozone_usage DAG demonstrates the basic Ozone workflow. It creates a volume and bucket, uploads a file to Ozone, and verifies that the uploaded object exists.

Example of creating a volume and a bucket:

from airflow.providers.arenadata.ozone.operators.ozone import (
    OzoneCreateBucketOperator,
    OzoneCreateVolumeOperator,
)

create_volume = OzoneCreateVolumeOperator(
    task_id="create_volume",
    volume_name="vol1", (1)
    quota="10GB", (2)
    ozone_conn_id="ozone_admin_default", (3)
)

create_bucket = OzoneCreateBucketOperator(
    task_id="create_bucket",
    volume_name="vol1", (4)
    bucket_name="bucket-native", (5)
    quota="10GB",
    ozone_conn_id="ozone_admin_default",
)
1 Name of the volume to create.
2 Volume storage quota.
3 Airflow connection used to communicate with Ozone.
4 Target volume.
5 Bucket created inside the specified volume.

Example of uploading a file to Ozone:

from airflow.providers.arenadata.ozone.operators.ozone import OzoneUploadContentOperator

upload_file = OzoneUploadContentOperator(
    task_id="upload_file",
    content="Hello from Airflow",
    remote_path="ofs://om-service/vol1/bucket-native/data_dir/file.txt", (1)
    ozone_conn_id="ozone_admin_default",
    if_exists="overwrite", (2)
)
1 Destination path in the Ozone filesystem.
2 Replaces the existing object, allowing the example DAG to be executed repeatedly.

Example of verifying that the uploaded file exists:

from airflow.providers.arenadata.ozone.sensors.ozone import OzoneKeySensor

wait_for_file = OzoneKeySensor(
    task_id="wait_for_file",
    path="ofs://om-service/vol1/bucket-native/data_dir/file.txt", (1)
    ozone_conn_id="ozone_admin_default",
    mode="reschedule", (2)
)
1 Path of the object to monitor.
2 Releases the worker slot between polling attempts.

Data lifecycle management

The example_ozone_data_lifecycle DAG demonstrates a complete data lifecycle workflow, including file discovery, processing, archiving, backup, and cleanup.

Example of listing files stored in a bucket:

from airflow.providers.arenadata.ozone.operators.ozone import OzoneListOperator

list_files = OzoneListOperator(
    task_id="list_files_in_landing_zone",
    path="ofs://om-service/landing/raw/*", (1)
    ozone_conn_id="ozone_admin_default",
)
1 Lists all objects located in the landing area. The result is stored in XCom for downstream tasks.

Example of creating an archive with processed files:

from airflow.providers.arenadata.ozone.operators.ozone import OzoneMoveOperator

archive_files = OzoneMoveOperator(
    task_id="archive_landing_files",
    source_path="ofs://om-service/landing/raw/*", (1)
    dest_path="ofs://om-service/archive/processed/ds={{ ds }}", (2)
    ozone_conn_id="ozone_admin_default",
)
1 Moves all processed files from the landing directory.
2 Stores archived data in a date-partitioned directory.

Example of deleting the original files:

from airflow.providers.arenadata.ozone.operators.ozone import OzoneDeleteKeyOperator

cleanup = OzoneDeleteKeyOperator(
    task_id="cleanup_landing_zone",
    path="ofs://om-service/landing/raw/*",
    ozone_conn_id="ozone_admin_default",
)

This workflow can be used as a template for automated archive and retention pipelines.

Administrative workflow

The example_ozone_multi_tenant_management DAG automates provisioning of storage resources for new projects or tenants.

Example of creating a project volume and configuring a storage quota:

from airflow.providers.arenadata.ozone.operators.ozone import (
    OzoneCreateVolumeOperator,
    OzoneSetQuotaOperator,
)

create_volume = OzoneCreateVolumeOperator(
    task_id="create_project_volume",
    volume_name="project-alpha", (1)
    ozone_conn_id="ozone_admin_default",
)

set_quota = OzoneSetQuotaOperator(
    task_id="set_project_quota",
    volume="project-alpha",
    quota="10GB", (2)
    ozone_conn_id="ozone_admin_default",
)
1 Creates an isolated storage volume for a project.
2 Limits the maximum storage available to the volume.

Example of creating a project bucket:

create_landing_bucket = OzoneCreateBucketOperator(
    task_id="create_landing_bucket",
    volume_name="project-alpha",
    bucket_name="landing", (1)
    quota="1GB",
    ozone_conn_id="ozone_admin_default",
)
1 Creates a bucket for project data within the provisioned volume.

Data transfer

The example_ozone_copy_from_hdfs DAG demonstrates how to migrate data from HDFS to Ozone using Hadoop DistCp.

The workflow performs the following operations:

  1. Waits for a trigger file in Ozone.

  2. Sets a quota on the destination bucket.

  3. Copies data from HDFS to Ozone.

For demonstration purposes, the DAG creates the trigger file automatically after a short delay. In production environments, this helper branch can be removed and the trigger file can be created by an upstream ingestion workflow.

Example of copying data from HDFS to Ozone:

from airflow.providers.arenadata.ozone.transfers.hdfs_to_ozone import HdfsToOzoneOperator

copy_hdfs_to_ozone = HdfsToOzoneOperator(
    task_id="copy_hdfs_to_ozone",
    source_path="hdfs:///user/data/legacy/", (1)
    dest_path="ofs://om-service/vol1/bucket1/migrated/", (2)
    hdfs_conn_id="hdfs_admin_default", (3)
)
1 Source directory in HDFS.
2 Destination directory in the Ozone filesystem.
3 Airflow connection used to access HDFS.

Monitoring

Use OzoneKeySensor to wait until an object becomes available in Ozone.

This sensor is commonly used to synchronize independent workflows, detect arrival of new datasets, or wait until upstream applications finish writing data.

from airflow.providers.arenadata.ozone.sensors.ozone import OzoneKeySensor

wait_for_trigger = OzoneKeySensor(
    task_id="wait_for_trigger_file",
    path="ofs://om-service/vol1/bucket1/trigger.lck", (1)
    ozone_conn_id="ozone_admin_default",
    mode="reschedule",
)
1 Object whose existence triggers execution of downstream tasks.

Backup creation

Example of creating a snapshot backup:

from airflow.providers.arenadata.ozone.transfers.ozone_backup import OzoneBackupOperator

backup_archive = OzoneBackupOperator(
    task_id="backup_archive_via_snapshot",
    volume="archive", (1)
    bucket="processed", (2)
    snapshot_name="snap-{{ ds_nodash }}", (3)
    ozone_conn_id="ozone_admin_default",
)
1 Volume containing archived data.
2 Bucket to snapshot.
3 Snapshot name generated from the configured prefix and DAG execution date.
Found a mistake? Seleсt text and press Ctrl+Enter to report it