I am writing a airflow task to read a large csv and save it to postgresql database. I found this asyncpg package that has a copy function which runs much faster than any other packages. However, it is async, and I don't know how to incorporate it into Airflow. Here is a sample code: from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime, timedelta from pandas import DataFrame import asyncpg async def to_sql(dataframe, table_name, schema_name='public', timeout=None, truncate=False): connection = await asyncpg.connect(user='postgres', host='host.docker.internal', database='quantaxis', password='123456') result = await connection.copy_records_to_table( table_name, records=dataframe.values.tolist(), columns=shared_columns, schema_name=schema_name, timeout=timeout) await connection.close() return result default_args = { 'owner': 'Airflow', 'depends_on_past': False, 'start_date': datetime(2020, 1, 1), 'retries': 1, 'retry_delay': timedelta(minutes=1), } dag = DAG('pythonexp2123', default_args=default_args, schedule_interval=timedelta(days=1)) async def save_file_to_database(ds): df = pd.read_csv("data{0}.csv".format(ds)) r = await to_sql(df, 'test') return r t1 = PythonOperator( task_id='pushing_task', provide_context=True, python_callable=save_file_to_database, dag=dag ) t1 When I run it, it will return error: Can't Pickle Object <Corountine> How could I change the function to make this Dag work? I still want to use asyncpg package because of its speed. Continue reading...
