将 Apache Airflow 与 Timescale Cloud 集成

启用 Airflow 和 Timescale Cloud 之间的 PostgreSQL 连接

pip install psycopg2-binary

复制

在 Airflow UI 中启用 PostgreSQL 连接类型

pip install apache-airflow-providers-postgres

复制

运行 Airflow

在您的开发机器上，运行以下命令：

airflow standalone

复制

Airflow UI 的用户名和密码显示在输出的 standalone | Login with username 行中。

从 Airflow 添加到您的 Timescale Cloud 服务的连接

1. 在您的浏览器中，导航到 localhost:8080，然后选择 Admin > Connections。
2. 点击 +（添加新记录），然后使用您的连接信息填写表格。Connection Type（连接类型）为 Postgres。

创建并执行 DAG

要从 Airflow 向您的 Timescale Cloud 服务插入数据：

1. 在 $AIRFLOW_HOME/dags/timescale_dag.py 中，添加以下代码：

   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.timescaledb.cn/getting-started/latest/try-key-features-timescale-products/#optimize-time-series-data-in-hypertables
        """            
       cursor.execute("INSERT INTO crypto_assets (symbol, name) VALUES (%s, %s)",
        ('NEW/Asset','New Asset 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,
   )

   复制

   此 DAG 使用在为关系数据创建常规 PostgreSQL 表中创建的 company 表。

2. 在您的浏览器中，刷新Airflow UI。

3. 在 Search DAGS（搜索 DAG）中，键入 timescale_dag 并按 Enter 键。

4. 点击播放图标并触发 DAG

   

   展开图片

验证数据是否出现在 Timescale Cloud 中

1. 在Timescale 控制台中，导航到您的服务并点击 SQL editor（SQL 编辑器）。

2. 运行查询以查看您的数据。例如：SELECT symbol, name FROM company;。

   您会看到表中插入的新行。