> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Airflow operators

Operators are one of the building blocks of Airflow DAGs. There are many different types of operators available in Airflow. The `PythonOperator` can execute any Python function, and is functionally equivalent to using the `@task` decorator, while other operators contain pre-created logic to perform a specific task, such as executing a Bash script (`BashOperator`) or running a SQL query in a relational database (`SQLExecuteQueryOperator`). Operators are used alongside other building blocks, such as [decorators](/docs/learn/airflow-decorators) and [hooks](/docs/learn/what-is-a-hook), to create tasks in a DAG written with the task-oriented approach. Operators classes can be imported from Airflow provider packages.

In this guide, you'll learn the basics of using operators in Airflow.

To view a list of available operators available in different Airflow provider packages, go to the [Airflow Registry](https://airflow.apache.org/registry).

## Assumed knowledge

To get the most out of this guide, you should have an understanding of:

* Basic Airflow concepts. See [Introduction to Apache Airflow](/docs/learn/intro-to-airflow).
* Basic Python. See the [Python Documentation](https://docs.python.org/3/tutorial/index.html).

## Operator basics

Operators are Python classes that encapsulate logic to do a unit of work. They can be viewed as a wrapper around each unit of work that defines the actions that will be completed and abstract the majority of code you would typically need to write. When you create an instance of an operator in a DAG and provide it with its required parameters, it becomes a task.

A base set of operators is contained in the [Airflow standard provider](https://airflow.apache.org/docs/apache-airflow-providers-standard/stable/index.html) package, which is pre-installed when using the Astro CLI. Other operators are contained in specialized provider packages, often centered around a specific technology or service. For example, the [Airflow Snowflake Provider](https://airflow.apache.org/registry/providers/snowflake/) package contains operators for interacting with Snowflake, while the [Airflow Google provider](https://airflow.apache.org/registry/providers/google/) package contains operators for interacting with Google Cloud services. There are also several packages that contain operators that can be used with a set of services:

* [Common SQL](https://airflow.apache.org/registry/providers/common-sql/)
* [Common IO](https://airflow.apache.org/registry/providers/common-io/)
* [Common Messaging](https://airflow.apache.org/docs/apache-airflow-providers-common-messaging/stable/index.html)

## Operator examples

Following are some of the most frequently used Airflow operators. Note that only a few of the possible parameters are shown, refer to the [Airflow Registry](https://airflow.apache.org/registry/) for a full list of parameters for each operator.

* [`PythonOperator`](https://airflow.apache.org/registry/providers/standard#standard-python-PythonOperator): Executes a Python function. It is functionally equivalent to using the `@task` decorator. See, [Introduction to the TaskFlow API and Airflow decorators](/docs/learn/airflow-decorators).

  ```python wrap theme={null}
  from airflow.providers.standard.operators.python import PythonOperator

  def _my_python_function():
      print("Hello world!")

  my_task = PythonOperator(
      task_id="my_task",
      python_callable=_my_python_function,
  )
  ```

* [`BashOperator`](https://airflow.apache.org/registry/providers/standard#standard-bash-BashOperator): Executes a bash script. See also the [Using the `BashOperator`](/docs/learn/bashoperator) guide.

  ```python wrap theme={null}
  from airflow.providers.standard.operators.bash import BashOperator

  my_task = BashOperator(
      task_id="my_task",
      bash_command="echo 'Hello world!'",
  )
  ```

* [`KubernetesPodOperator`](https://airflow.apache.org/registry/providers/cncf-kubernetes#cncf-kubernetes-pod-KubernetesPodOperator): Executes a task defined as a Docker image in a Kubernetes Pod. See, [Use the `KubernetesPodOperator`](/docs/learn/kubepod-operator).

  ```python wrap theme={null}
  from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

  my_task = KubernetesPodOperator(
      task_id="my_task",
      kubernetes_conn_id="<my-kubernetes-connection>",
      name="<my-pod-name>",
      namespace="<my-namespace>",
      image="python:3.12-slim", # Docker image to run
      cmds=["python", "-c"], # Command to run in the container
      arguments=["print('Hello world!')"], # Arguments to the command
  )
  ```

* [`SQLExecuteQueryOperator`](https://airflow.apache.org/registry/providers/common-sql#common-sql-sql-SQLExecuteQueryOperator): Executes a SQL query against a relational database.

  ```python wrap theme={null}
  from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

  my_task = SQLExecuteQueryOperator(
      task_id="my_task",
      sql="SELECT * FROM my_table",
      database="<my-database>",
      conn_id="<my-connection>",
  )
  ```

* [`EmptyOperator`](https://airflow.apache.org/registry/providers/standard#standard-empty-EmptyOperator): A no-op operator that does nothing. This is useful for creating placeholder tasks in a DAG.

  ```python wrap theme={null}
  from airflow.providers.standard.operators.empty import EmptyOperator

  my_task = EmptyOperator(task_id="my_task")
  ```

All operators inherit from the abstract [`BaseOperator` class](https://github.com/apache/airflow/blob/main/task-sdk/src/airflow/sdk/bases/operator.py), which contains the logic to execute the work of the operator within the context of a DAG.

Arguments of the `BaseOperator` class can be passed to all operators. The most common arguments are:

* `task_id`: A unique identifier for the task. This is required for all operators.
* `retries`: The number of times to retry the task if it fails. This is optional and defaults to 0. See [Rerun Airflow DAGs and tasks](/docs/learn/rerunning-dags#automatically-retry-tasks).
* `pool`: The name of the pool to use for the task. This is optional and defaults to None. See [Airflow pools](/docs/learn/airflow-pools).
* `execution_timeout`: The maximum time to wait for the task to complete. This is optional and defaults to None. It is a good practice to set this value to prevent tasks from running indefinitely.

You can set these arguments and other `BaseOperator` arguments (other than `task_id` which needs to be unique per operator) at the DAG level for all tasks in a DAG. By using the `default_args` dictionary. You can override these values for individual tasks by setting the same arguments in the task definition.

```python expandable wrap theme={null}
import hashlib
import json

from airflow.exceptions import AirflowException
from airflow.decorators import dag, task
from airflow.models import Variable
from airflow.models.baseoperator import chain
from airflow.operators.empty import EmptyOperator
from airflow.utils.dates import datetime
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.amazon.aws.transfers.local_to_s3 import (
    LocalFilesystemToS3Operator,
)
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.operators.sql import SQLCheckOperator
from airflow.utils.task_group import TaskGroup


# The file(s) to upload shouldn't be hardcoded in a production setting,
# this is just for demo purposes.
CSV_FILE_NAME = "forestfires.csv"
CSV_FILE_PATH = f"include/sample_data/forestfire_data/{CSV_FILE_NAME}"


@dag(
    "simple_redshift_3",
    start_date=datetime(2021, 7, 7),
    description="""A sample Airflow DAG to load data from csv files to S3
                 and then Redshift, with data integrity and quality checks.""",
    schedule=None,
    template_searchpath="/usr/local/airflow/include/sql/redshift_examples/",
    catchup=False,
)
def simple_redshift_3():
    """
    Before running the DAG, set the following in an Airflow
    or Environment Variable:
    - key: aws_configs
    - value: { "s3_bucket": [bucket_name], "s3_key_prefix": [key_prefix],
             "redshift_table": [table_name]}
    Fully replacing [bucket_name], [key_prefix], and [table_name].
    """

    upload_file = LocalFilesystemToS3Operator(
        task_id="upload_to_s3",
        filename=CSV_FILE_PATH,
        dest_key="{{ var.json.aws_configs.s3_key_prefix }}/" + CSV_FILE_PATH,
        dest_bucket="{{ var.json.aws_configs.s3_bucket }}",
        aws_conn_id="aws_default",
        replace=True,
    )

    @task
    def validate_etag():
        """
        #### Validation task
        Check the destination ETag against the local MD5 hash to ensure
        the file was uploaded without errors.
        """
        s3 = S3Hook()
        aws_configs = Variable.get("aws_configs", deserialize_json=True)
        obj = s3.get_key(
            key=f"{aws_configs.get('s3_key_prefix')}/{CSV_FILE_PATH}",
            bucket_name=aws_configs.get("s3_bucket"),
        )
        obj_etag = obj.e_tag.strip('"')
        # Change `CSV_FILE_PATH` to `CSV_CORRUPT_FILE_PATH` for the "sad path".
        file_hash = hashlib.md5(open(CSV_FILE_PATH).read().encode("utf-8")).hexdigest()
        if obj_etag != file_hash:
            raise AirflowException(
                """Upload Error: Object ETag in S3 did not match
                hash of local file."""
            )

    # Tasks that were created using decorators have to be called to be used
    validate_file = validate_etag()

    # --- Create Redshift Table --- #
    create_redshift_table = PostgresOperator(
        task_id="create_table",
        sql="create_redshift_forestfire_table.sql",
        postgres_conn_id="redshift_default",
    )

    # --- Second load task --- #
    load_to_redshift = S3ToRedshiftOperator(
        task_id="load_to_redshift",
        s3_bucket="{{ var.json.aws_configs.s3_bucket }}",
        s3_key="{{ var.json.aws_configs.s3_key_prefix }}" + f"/{CSV_FILE_PATH}",
        schema="PUBLIC",
        table="{{ var.json.aws_configs.redshift_table }}",
        copy_options=["csv"],
    )

    # --- Redshift row validation task --- #
    validate_redshift = SQLCheckOperator(
        task_id="validate_redshift",
        conn_id="redshift_default",
        sql="validate_redshift_forestfire_load.sql",
        params={"filename": CSV_FILE_NAME},
    )

    # --- Row-level data quality check --- #
    with open("include/validation/forestfire_validation.json") as ffv:
        with TaskGroup(group_id="row_quality_checks") as quality_check_group:
            ffv_json = json.load(ffv)
            for id, values in ffv_json.items():
                values["id"] = id
                SQLCheckOperator(
                    task_id=f"forestfire_row_quality_check_{id}",
                    conn_id="redshift_default",
                    sql="row_quality_redshift_forestfire_check.sql",
                    params=values,
                )

    # --- Drop Redshift table --- #
    drop_redshift_table = PostgresOperator(
        task_id="drop_table",
        sql="drop_redshift_forestfire_table.sql",
        postgres_conn_id="redshift_default",
    )

    begin = EmptyOperator(task_id="begin")
    end = EmptyOperator(task_id="end")

    # --- Define task dependencies --- #
    chain(
        begin,
        upload_file,
        validate_file,
        create_redshift_table,
        load_to_redshift,
        validate_redshift,
        quality_check_group,
        drop_redshift_table,
        end,
    )


simple_redshift_3()
```

## Best practices

Operators typically only require a few parameters. Keep the following considerations in mind when using Airflow operators:

* The [Airflow Registry](https://airflow.apache.org/registry) is the best resource for learning what operators are available and how they are used.
* The [Airflow standard provider](https://airflow.apache.org/docs/apache-airflow-providers-standard/stable/index.html) package includes basic operators such as the `PythonOperator` and `BashOperator`. These operators are automatically available in your Airflow environment if you are using the Astro CLI. All other operators are part of provider packages, some which you must install separately, depending on what type of Airflow distribution you are using.
* You can combine operators and [decorators](/docs/learn/airflow-decorators) freely in the same DAG. Many users choose to use the `@task` decorator for most of their tasks, and add operators for tasks where a specialized operator exists for their use case. The example above shows a DAG with one operator (`BashOperator`) and one `@task` decorated task.
* If an operator exists for your specific use case, you should use it instead of your own Python functions or [hooks](/docs/learn/what-is-a-hook). This makes your DAGs easier to read and maintain.
* If an operator doesn't exist for your use case, you can either use custom Python code in an `@task` decorated task or `PythonOperator` or extend an operator to meet your needs. For more information about customizing operators, see [Custom hooks and operators](/docs/learn/airflow-importing-custom-hooks-operators).
* [Sensors](/docs/learn/what-is-a-sensor) are a type of operator that waits for something to happen. They can be used to detect events in systems outside of Airflow.
* [Deferrable Operators](/docs/learn/deferrable-operators) are a type of operator that releases their worker slot while waiting for their work to be completed. This can result in cost savings and greater scalability. Astronomer recommends using deferrable operators whenever one exists for your use case and your task takes longer than a minute. A lot of operators that potentially need to wait for something have a deferrable mode which you can enable by setting their `deferrable` parameter to `True`.
* Any operator that interacts with a service external to Airflow typically requires a connection so that Airflow can authenticate to that external system. For more information about setting up connections, see [Managing your connections in Apache Airflow](/docs/learn/connections) or in the examples to follow.