我想使用 Airflow 的 SimpleHttpOperator 通過 REST API 上傳二進制資料。
對于 python 的請求，我會這樣做
upload_response = requests.put(
upload_url, data=Path(f"{filename}").read_bytes(), headers=headers
)
在 Airflow 中，我的第一個任務是使用 SFTPOperator 下載檔案，然后嘗試使用 SimpleHttpOperator 上傳它：
get_file = SFTPOperator(
task_id = 'get_file',
ssh_conn_id = 'my_sftp_conn',
remote_filepath = f"{templ_remote_filepath}/{filename}",
local_filepath = f"{basepath}/{filename}",
operation = 'GET',
)
upload_file = SimpleHttpOperator(
task_id = 'upload_file',
http_conn_id = "my_http_conn",
endpoint = get_upload_url.output,
method = 'PUT',
headers = {"Content-MD5": md5_hash_b64, "Content-Type": "application/text"},
data = Path(get_file.output).read_bytes(),
log_response = True,
)
我希望能夠輸入get_file.output（下載的檔案路徑）作為字串而不是 XComArgs 原樣。我怎樣才能做到這一點？
uj5u.com熱心網友回復：
在這種情況下，您不應使用運算子。每個操作員將運行自己的“任務實體”，因此它們可能運行在完全不同的實體上。這就是操作員使用 XCom 來存盤資料的原因，因為任務之間沒有其他通信方式。默認情況下，它的大小相當有限，因為它使用資料庫（在您的情況下，它將在一個任務中將檔案保存到資料庫中，并在另一個任務中檢索它 - 這幾乎肯定不是您想要的）。
您應該做的是，您應該使用單個 PythonOperator 和 HOOKS 而不是運算子。Airflow“API”有兩個級別 - 操作員在那里提供可以用作完整任務的“塊”，但實際上它們通常是圍繞 Hook 的薄包裝，提供相同的功能，但可以組合使用 - 可以使用多個鉤子在同一個運營商。
定義自定義 PythonOperator 非常簡單 - 特別是如果您使用 TaskFlow API 的現代方法。
請參閱教程https://airflow.apache.org/docs/apache-airflow/stable/tutorial_taskflow_api.html
所以在你的情況下，你應該有類似的東西（它不太可能編譯，但會給你靈感）：
@task()
def transfer():
tempfile.TemporaryDirectory() as tmpfile:
sftp_hook = SFTPHook(ssh_conn_id = 'my_sftp_conn')
sftp_hook.retrieve_file(remote_file_name='REMOTE_URL_HERE', local_filename=tmp_file)
data = Path(tmp_file).read_bytes()
http_hook = HTTPHook()
http_hook.run(data=data,.....)
然而，將資料加載到記憶體并不是超級優化的，你甚至可以直接使用請求庫而不是 HttpHook 來上傳檔案（或者甚至 HttpHook 已經實作了流。
兩個鉤子的完整檔案：
HtpHook：https ://airflow.apache.org/docs/apache-airflow-providers-http/stable/_api/airflow/providers/http/hooks/http/index.html
SFTPHook：https ://airflow.apache.org/docs/apache-airflow-providers-sftp/stable/_api/airflow/providers/sftp/hooks/sftp/index.html
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/371905.html
上一篇：為什么OpenAPI(JSONSchema)不支持無符號整數？
下一篇：測驗彈簧控制器時出現意外回傳碼