气流-脚本在初始化期间执行

我有一个Airflow脚本,可以将数据从表2插入表1。作为airflow初始化过程的一部分,我看到insert函数一直在后台运行,即使我没有触发它或计划它。我想知道脚本中有什么错误导致它自动触发。我需要在下面的脚本中修改什么,以确保它不会在初始化过程中执行该命令。

## Library Imports
import psycopg2
import airflow
from airflow import DAG
from airflow.operators import BashOperator
from sqlalchemy import create_engine
import io


# Following are defaults which can be overridden later on
default_args = {
'owner': 'admin',
'depends_on_past': False,
'start_date': datetime(2018, 5, 25),
'email': ['admin@mail.com'],
'email_on_failure': True,
'email_on_retry': True,
'retries': 1,
'retry_delay': timedelta(minutes=1),
}

dag = DAG('sample', default_args=default_args)


#######################

def db_login():
    global db_con
try:
    db_con = psycopg2.connect(" dbname = 'db' user = 'user' password = 'pass' host = 'hostname' port = '5439' sslmode = 'require' ")
except:
    print("I am unable to connect to the database.")
print('Connection Task Complete: Connected to DB')
return(db_con)


#######################

def insert_data():
    cur = db_con.cursor()
    cur.execute("""insert into table_1 select id,name,status from table_2 limit 2 ;""")
    db_con.commit()
    print('ETL Task Complete: Inserting data into table_1')


db_login()
insert_data()
db_con.close()

##########################################


t1 = BashOperator(
task_id='db_con',
python_callable=db_login(),
bash_command='python3 ~/airflow/dags/sample.py',
email_on_failure=True,
email=['admin@mail.com'],
dag=dag)

t2 = BashOperator(
task_id='insert',
python_callable=insert_data(),
bash_command='python3 ~/airflow/dags/sample.py',
email_on_failure=True,
email=['admin@mail.com'],
dag=dag)


t1.set_downstream(t2)

有人能帮忙吗。谢谢

更新代码:

## Third party Library Imports

import psycopg2
import airflow
from airflow import DAG
from airflow.operators import BashOperator
from datetime import datetime, timedelta
from sqlalchemy import create_engine
import io



default_args = {
'owner': 'admin',
#'depends_on_past': False,
'start_date': datetime(2018, 5, 25),
 'email': ['admin@mail.com'],
 'email_on_failure': True,
 'email_on_retry': True,
 'retries': 1,
 'retry_delay': timedelta(minutes=1), }

dag = DAG('sample', default_args=default_args, catchup=False, schedule_interval="@once")


def db_login():
    global db_con
    try:
        db_con = psycopg2.connect(
        " dbname = 'db' user = 'user' password = 'password' host = 'host' port = '5439' sslmode = 'require' ")
    except:
        print("I am unable to connect to the database.")
    print('Connection success')
    return (db_con)

def insert_data():
    cur = db_con.cursor()
    cur.execute("""insert into table_1 select id,name,status from table_2 limit 2;""")
    db_con.commit()
    print('ETL Task Complete: Inserting data into table_1')

def load_etl():
    db_login()
    insert_data()
    dwh_connection.close()

#Function to execute the query
load_etl()

t1 = BashOperator(
    task_id='db_connection',
    python_callable=load_etl(),
    bash_command='python3 ~/airflow/dags/sample.py',
    email_on_failure=True,
    email=['admin@mail.com'],
    dag=dag)

#t2 = BashOperator(
#task_id='ops_load_del',
#python_callable=insert_data(),
#bash_command='python3 ~/airflow/dags/sample.py',
#email_on_failure=True,
#email=['admin@mail.com'],
#dag=dag)

t1
#t1.set_downstream(t2)