通过注意到SFTP https://airflow.apache.org/docs/stable/_modules/airflow/contrib/operators/sftp_operator.html操作员使用 ssh_hook 打开 sftp 传输通道,您应该需要提供ssh_hook
or ssh_conn_id
用于文件传输。首先,让我们看一个提供参数的示例ssh_conn_id
.
from airflow.providers.sftp.operators import sftp_operator
from airflow import DAG
import datetime
dag = DAG(
'test_dag',
start_date = datetime.datetime(2020,1,8,0,0,0),
schedule_interval = '@daily'
)
put_operation = SFTPOperator(
task_id="operation",
ssh_conn_id="ssh_default",
local_filepath="route_to_local_file",
remote_filepath="remote_route_to_copy",
operation="put",
dag=dag
)
get_operation = SFTPOperator(....,
operation = "get",
dag = dag
)
put_operation >> get_operation
请注意,应根据任务的需要安排 dag,此处的示例考虑从中午开始的每日计划。现在,如果您提供 SSHhook,则需要对上述代码进行以下更改
from airflow.contrib.hooks.ssh_hook import SSHHook
...
put_operation = SFTPOperator(
task_id="operation",
ssh_hook=SSHHook("Name_of_variable_defined"),
...
dag=dag
)
....
where "Name_of_variable_defined"
在Airflow界面的Admin -> Connections中创建。