azure - SFTP 到 Azure Blob 存储

Question

我正在尝试使用 SFTPToWasbOperator 将文件从 SFTP 复制到 Azure Blob 存储。我收到错误。好像我做错了什么，但我不知道它是什么。有人可以检查一下下面的代码并看看它是否有问题吗？

Airflow 日志

**[2022-07-10, 13:08:48 UTC] {sftp_to_wasb.py:188} 信息 - 将/SPi_ESG_Live/07-04-2022/DataPoint_2022_07_04.csv 上传到 wasb://testcotainer 作为 https://test.blob.core.windows.net/testcotainer/DataPoint_2022_07_04.csv

[2022-07-10, 13:08:48 UTC] {_universal.py:473} 信息 - 请求 URL:' https://.blob.core.windows.net/***/test/https%3A//test.blob.core.windows.net/testcontainer/DataPoint_2022_07_04.csv '

错误消息

“azure.core.exceptions.ServiceRequestError:URL 具有无效标签。”

Airflow DAG

import os
from datetime import datetime

from airflow import DAG
from airflow.decorators import task
from airflow.providers.microsoft.azure.operators.wasb_delete_blob import WasbDeleteBlobOperator
from airflow.providers.microsoft.azure.transfers.sftp_to_wasb import SFTPToWasbOperator
from airflow.providers.sftp.hooks.sftp import SFTPHook
from airflow.providers.sftp.operators.sftp import SFTPOperator

AZURE_CONTAINER_NAME = "testcotainer"
BLOB_PREFIX = "https://test.blob.core.windows.net/testcotainer/"
SFTP_SRC_PATH = "/SPi_test_Live/07-04-2022/"
ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
DAG_ID = "example_sftp_to_wasb"


with DAG(
    DAG_ID,
    schedule_interval=None,
    catchup=False,
    start_date=datetime(2021, 1, 1),  # Override to match your needs
) as dag:
    # [START how_to_sftp_to_wasb]
    transfer_files_to_azure = SFTPToWasbOperator(
        task_id="transfer_files_from_sftp_to_wasb",
        # SFTP args
        sftp_source_path=SFTP_SRC_PATH,
        # AZURE args
        container_name=AZURE_CONTAINER_NAME,
        blob_prefix=BLOB_PREFIX,
    )
    # [END how_to_sftp_to_wasb]

Answer 1

问题出在 BLOB_PREFIX 上，它不是一个 url，而是 azure url 后面的前缀

请参阅此源示例:https://airflow.apache.org/docs/apache-airflow-providers-microsoft-azure/stable/_modules/tests/system/providers/microsoft/azure/example_sftp_to_wasb.html