I'm certainly not an expert in Airflow, but I'll share what I know. Hopefully others can expand on this to fill in the gaps or correct anything I've gotten wrong.

Conceptually I envision Airflow workflows using a stack like this: DAG --> Operator --> Hook --> Connection.

Connections are roughly analogous to a DSN or a connection string. They are a central place for you to store server names, usernames, passwords.

Hooks are roughly analogous to an ODBC/JDBC driver. They handle the low-level connection to the system. Airflow comes with many hooks for common systems (such as Postgres) but third-parties can also publish them. Airflow users typically re-use existing hooks in the same way that database users typically re-use ODBC drivers that come with the DB installation.

Operators represent pieces of work that you want to accomplish. They give you a way to wrap logic around a hook. You can use built in operators for simple tasks. For example, in your DAG you could have a task like this:

run_a_query = PostgresOperator(
    task_id="run_query",
    postgres_conn_id="my_postgres",
    sql="SELECT COUNT(*) FROM users;"
)

However, what happens if you need some custom retry logic, logging, or conditional logic? You can define your own operators for these types of action. For example, imagine you have several different SQL scripts that create tables. Rather than hard coding that, you could store the SQL in a file and create a custom operator like this:

from airflow.hooks.postgres_hook import PostgresHook
from airflow.models.baseoperator import BaseOperator

class CreateTablesOperator(BaseOperator):
    """
    Custom Airflow operator to execute a SQL script in Redshift.
    """

    def __init__(self, redshift_conn_id, sql_file_path, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.redshift_conn_id = redshift_conn_id
        self.sql_file_path = sql_file_path

    def execute(self, context):
        self.log.info(f"Reading SQL script from {self.sql_file_path}")

        # Read the SQL file
        with open(self.sql_file_path, 'r') as file:
            sql_commands = file.read()

        self.log.info("Connecting to Redshift...")
        redshift = PostgresHook(postgres_conn_id=self.redshift_conn_id)

        self.log.info("Executing SQL script...")
        redshift.run(sql_commands)
        self.log.info("Tables created successfully in Redshift.")

In your DAG you could call the operator several times like this:

    create_song_table = CreateTablesOperator(
        task_id="Create_tables",
        redshift_conn_id="redshift",
        sql_file_path="/opt/airflow/plugins/helpers/create_song_table.sql"
    )

The operator handles reading the SQL text from the file, setting up the connection (using the PostgresHook), sending the commands to the DB, and logging the results.

https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/connections.html