Integrate Apache Airflow with Timescale Cloud

Enable PostgreSQL connections between Airflow and Timescale Cloud

pip install psycopg2-binary

Copy

Enable PostgreSQL connection types in the Airflow UI

pip install apache-airflow-providers-postgres

Copy

Run Airflow

On your development machine, run the following command:

airflow standalone

Copy

The username and password for Airflow UI are displayed in the standalone | Login with username line in the output.

Add a connection from Airflow to your Timescale Cloud service

1. In your browser, navigate to localhost:8080, then select Admin > Connections.
2. Click + (Add a new record), then use your connection info to fill in the form. The Connection Type is Postgres.

Create and execute a DAG

To insert data in your Timescale Cloud service from Airflow:

1. In $AIRFLOW_HOME/dags/timescale_dag.py, add the following code:

   from airflow import DAG
   from airflow.operators.python_operator import PythonOperator
   from airflow.hooks.postgres_hook import PostgresHook
   from datetime import datetime
   
   def insert_data_to_timescale():
       hook = PostgresHook(postgres_conn_id='the ID of the connenction you created')
       conn = hook.get_conn()
       cursor = conn.cursor()
       """
         This could be any query. This example inserts data into the table
         you create in:
   
         https://docs.timescale.com/getting-started/latest/tables-hypertables/#create-regular-postgresql-tables-for-relational-data
        """            
       cursor.execute("INSERT INTO company (symbol, name) VALUES (%s, %s)",
               ('new_company_symbol', 'New Company Name'))
       conn.commit()
       cursor.close()
       conn.close()
   
   default_args = {
       'owner': 'airflow',
       'start_date': datetime(2023, 1, 1),
       'retries': 1,
   }
   
   dag = DAG('timescale_dag', default_args=default_args, schedule_interval='@daily')
   
   insert_task = PythonOperator(
       task_id='insert_data',
       python_callable=insert_data_to_timescale,
       dag=dag,
   )

   Copy

   This DAG uses the company table created in Create regular PostgreSQL tables for relational data.

2. In your browser, refresh the Airflow UI.

3. In Search DAGS, type timescale_dag and press ENTER.

4. Press the play icon and trigger the DAG:

   

   Expand image

Verify that the data appears in Timescale Cloud

1. In Timescale Console, navigate to your service and click SQL editor.

2. Run a query to view your data. For example: SELECT symbol, name FROM company;.

   You see the new rows inserted in the table.