为啥使用气流执行创建表语句时我的表不显示？

Posted 2023-03-31

【中文标题】为啥使用气流执行创建表语句时我的表不显示？

【英文标题】：Why isn't my table showing when doing a create table statement using airflow?为什么使用气流执行创建表语句时我的表不显示？

【发布时间】：2019-12-10 17:22:45

【问题描述】：

所以，我正在尝试使用气流在我的 Redshift DB 中创建一个表。我的连接有效，我使用 SQL 命令对其进行了测试，但是当我将 sql 命令更改为 create table 命令时，它成功运行，但它没有显示在我的 redshift DB 中。

这是我的代码：

from datetime import datetime, timedelta
from airflow import DAG
from airflow.hooks.postgres_hook import PostgresHook
from airflow.models import BaseOperator
from airflow.operators.python_operator import PythonOperator

default_args = 
    'owner': 'james_c',
    'depends_on_past': False,
    'start_date': datetime(2019,4,1),
    'email': ['[e-mail]'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=1)



def get_activated_sources():
    request ="CREATE TABLE if not exists schema1.db1.tb1 (vendor_id varchar(50) PRIMARY KEY, vendor_name VARCHAR(255) NOT NULL);"
    pg_hook = PostgresHook(postgre_conn_id="postgres_default", schema='schema1')
    connection = pg_hook.get_conn()
    cursor = connection.cursor()
    cursor.execute(request)
        #cursor.fetchall()
    cursor.close()
    connection.close()


with DAG('create_sample_table_dagg', description='testing my redshift connection', default_args=default_args, schedule_interval='@once', catchup=False) as dag:


    hook_task = PythonOperator(task_id='hook_task', python_callable=get_activated_sources)

关于为什么它没有在 redshift 中实际创建表而运行和完成的任何想法/建议？f

【问题讨论】：

这个问题可能重复吗？ dba.stackexchange.com/questions/94096/…

您是否尝试在 hook_task 下游添加 SQLSensor 以检查表是否存在？现在在您的代码中，没有任何内容检查表是否存在。此外，create 语句可以直接使用 PostgresOperator 而不是 PythonOperator 运行

【参考方案1】：

你的代码很好，你只需要写：

connection.commit()

下

cursor.execute(request)