Как запустить асинхронную функцию в воздушном потоке? ⇐ Python

[Anonymous]

Я пишу задачу воздушного потока, чтобы прочитать большой CSV и сохранить его в базе данных PostgreSQL.
Я нашел этот пакет AsyncpG, в котором функция копирования работает намного быстрее, чем любые другие пакеты. Тем не менее, это асинхронно, и я не знаю, как включить его в воздушный поток.
Вот пример кода: < /p>

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
< /code>

Когда я его запускаю, он вернет ошибку: < /p>

Can't Pickle Object

< /code>

Как я могу изменить функцию, чтобы этот DAG работал? Я все еще хочу использовать пакет Asyncpg из -за его скорости.


Подробнее здесь: https://stackoverflow.com/questions/602 ... in-airflow