我正在尝试为Airflow中创建的外部连接寻找连接池管理方法。
Airflow版本:2.1.0
Python版本:3.9.5
Airflow数据库:SQLite
已创建的外部连接:MySQL和Snowflake
我知道在airflow.cfg文件中有一些属性。
sql_alchemy_pool_enabled = True
sql_alchemy_pool_size = 5
但是这些属性是用于管理内部数据库的 airflow,而在我的情况下是 SQLite。
我有一些任务需要读取或写入 MySQL 和 Snowflake 中的数据。
snowflake_insert = SnowflakeOperator(
task_id='insert_snowflake',
dag=dag,
snowflake_conn_id=SNOWFLAKE_CONN_ID,
sql="Some Insert query",
warehouse=SNOWFLAKE_WAREHOUSE,
database=SNOWFLAKE_DATABASE,
schema=SNOWFLAKE_SCHEMA,
role=SNOWFLAKE_ROLE
)
并且
insert_mysql_task = MySqlOperator(task_id='insert_record', mysql_conn_id='mysql_default', sql="some insert query", dag=dag)
从MySQL读取数据
def get_records():
mysql_hook = MySqlHook(mysql_conn_id="mysql_default")
records = mysql_hook.get_records(sql=r"""Some select query""")
print(records)
我观察到在Snowflake中,每个任务(同一个dag中有多个任务)都会创建一个新的会话,对于MySQL尚未进行验证。
是否有一种方法可以为外部连接(在我的情况下是Snowflake和MySQL)维护连接池,或者通过其他方式在同一个会话中运行同一DAG中的所有查询?
谢谢