使用Apache气流检查文件是否存在于Azure DataLake上的最佳方法是什么？

Question

我有一个DAG，该DAG应检查是否已将文件上传到特定目录中的Azure Datalake。如果是这样，它允许其他DAG运行。

我考虑过使用FileSensor，但我认为FSCONNID参数不足以针对数据验证

Answer 1

Azure provider but you can easily implement one since the

我没有测试它，但这应该可以工作：

from airflow.providers.microsoft.azure.hooks.data_lake import AzureDataLakeHook
from airflow.sensors.base import BaseSensorOperator

class MyAzureDataLakeSensor(BaseSensorOperator):
    """
    Sense for files in Azure Data Lake

    :param path: The Azure Data Lake path to find the objects. Supports glob
        strings (templated)
    :param azure_data_lake_conn_id: The Azure Data Lake conn
    """

    template_fields: Sequence[str] = ('path',)
    ui_color = '#901dd2'

    def __init__(
        self, *, path: str, azure_data_lake_conn_id: str = 'azure_data_lake_default', **kwargs
    ) -> None:
        super().__init__(**kwargs)
        self.path = path
        self.azure_data_lake_conn_id = azure_data_lake_conn_id

    def poke(self, context: "Context") -> bool:
        hook = AzureDataLakeHook(azure_data_lake_conn_id=self.azure_data_lake_conn_id)
        self.log.info('Poking for file in path: %s', self.path)
        try:
            hook.check_for_file(file_path=self.path)
            return True
        except FileNotFoundError:
            pass
        return False

用法示例：

MyAzureDataLakeSensor(
    task_id='adls_sense',
    path='folder/file.csv',
    azure_data_lake_conn_id='azure_data_lake_default',
    mode='reschedule'
)

Answer 2

首先，看看。

我们可以看到，有专门的操作员对 Azure DataLake存储不幸的是，目前似乎只有 ADLSDELETEEPERATOR 可用。

此 adlsdeleteOperator 使用 azuredatalakehook 您应该在自己的自定义操作员中重复使用以检查文件是否存在。

我对您的建议是创建一个 checkoperator 使用ADLS钩检查输入中提供的文件是否存在 check_for_file hook的函数。

更新：正如注释中指出的那样，Checkoperator似乎是通过与SQL查询绑定的，并已弃用。使用自己的自定义传感器或自定义操作员是必经之路。

Answer 3

我使用拟议的API遇到了严重的问题。因此，我将Microsoft API嵌入了气流中。这很好。然后，您需要做的就是使用此操作员并通过Account_url和Access_Token。

from azure.storage.filedatalake import DataLakeServiceClient
from airflow.sensors.base import BaseSensorOperator

class AzureDataLakeSensor(BaseSensorOperator):

   def __init__(self, path, filename, account_url, access_token, **kwargs):
      super().__init__(**kwargs)
      self._client = DataLakeServiceClient(
            account_url=account_url,
            credential=access_token
      )

      self.path = path
      self.filename = filename

  def poke(self, context):
      container = self._client.get_file_system_client(file_system="raw")
      dir_client = container.get_directory_client(self.path)
      file = dir_client.get_file_client(self.filename)
      return file.exists()