如何在apache气流中使用bulk_load

Question

我有 apache airflow 2.1.4 和 postgres 数据库。 我需要一次插入多行。所以我打算使用 PostgresHook 的 bulk_load 方法，但每次都会出错。

data = pd.read_csv(open(filepath, 'rb'))
buffer = StringIO()
buffer.write(data.to_csv(index=None, header=None, sep='\t'))
buffer.seek(0)
schema_table = 'schema.table'
with PostgresHook(postgres_conn_id='my_pg_database'):
    PostgresHook.bulk_load(table=schema_table, tmp_file=buffer)

我得到的错误：

Traceback (most recent call last):
File "/home/airflow/dags/my_python_file.py", line 76, in <module>
my_func(filepath=my_file, target_schema=schema, target_table=table)
File "/home/airflow/dags/my_python_file.py", line 39, in my_func
with PostgresHook(postgres_conn_id='my_pg_database'):
AttributeError: __enter__

我什至找不到一些使用 bulk_load 的例子。将appriciate任何线索。谢谢。

Answer 1

Postgres Hook（以及任何其他真正的钩子）不是“上下文管理器”。您不能使用with: 来使用它们。

类似的东西应该可以工作：

postgres_hook = PostgresHook(postgres_conn_id='my_pg_database')
postgres_hook.bulk_load(...)