airflow.providers.microsoft.azure.hooks.data_lake¶

属性¶

凭据

类¶

`AzureDataLakeHook`	与 Azure Data Lake 的集成。
`AzureDataLakeStorageV2Hook`	与 ADLS gen2 存储账户交互。

模块内容¶

airflow.providers.microsoft.azure.hooks.data_lake.Credentials[source]¶

class airflow.providers.microsoft.azure.hooks.data_lake.AzureDataLakeHook(azure_data_lake_conn_id=default_conn_name)[source]¶

基类: airflow.hooks.base.BaseHook

与 Azure Data Lake 的集成。

AzureDataLakeHook 通过与 WebHDFS 兼容的 REST API 进行通信。请确保存在类型为 azure_data_lake 的 Airflow 连接。授权可以通过提供 *登录名* (=Client ID)，*密码* (=Client Secret)，以及 extra 字段 *tenant* (Tenant) 和 *account_name* (Account Name) 来完成。请参阅连接 azure_data_lake_default 以获取示例。

Client ID 和 secret 应作为用户和密码参数。Tenant 和 account name 应作为 extra 字段，格式为 {"tenant": "<TENANT>", "account_name": "ACCOUNT_NAME"}。

参数:: azure_data_lake_conn_id (str) – 引用 Azure Data Lake 连接。

conn_name_attr = 'azure_data_lake_conn_id'[source]¶

default_conn_name = 'azure_data_lake_default'[source]¶

conn_type = 'azure_data_lake'[source]¶

hook_name = 'Azure Data Lake'[source]¶

classmethod get_connection_form_widgets()[source]¶

返回要添加到连接表单的连接小部件。

classmethod get_ui_field_behaviour()[source]¶

返回自定义字段行为。

conn_id = 'azure_data_lake_default'[source]¶

account_name: str | None = None[source]¶

get_conn()[source]¶

返回一个 AzureDLFileSystem 对象。

check_for_file(file_path)[source]¶

检查 Azure Data Lake 上是否存在文件。

参数:: file_path (str) – 文件的路径和名称。
返回:: 如果文件存在则为 True，否则为 False。
返回类型:: bool

upload_file(local_path, remote_path, nthreads=64, overwrite=True, buffersize=4194304, blocksize=4194304, **kwargs)[source]¶

将文件上传到 Azure Data Lake。

参数:

local_path (str) – 本地路径。可以是单个文件、目录（此时会递归上传）或 glob 模式。不支持使用 ** 的递归 glob 模式。
remote_path (str) – 远程上传路径；如果文件较多，这是写入时的根目录。
nthreads (int) – 使用的线程数。如果为 None，则使用 CPU 核数。
overwrite (bool) – 是否强制覆盖现有文件/目录。如果为 False 且远程路径为目录，则无论是否有文件会被覆盖都会退出。如果为 True，则只覆盖匹配的文件名。
buffersize (int) – int [2**22] 内部缓冲区的字节数。此块不能大于一个 chunk，也不能小于一个 block。
blocksize (int) – int [2**22] 一个块的字节数。在每个 chunk 中，我们为每个 API 调用写入一个较小的块。此块不能大于一个 chunk。

download_file(local_path, remote_path, nthreads=64, overwrite=True, buffersize=4194304, blocksize=4194304, **kwargs)[source]¶

从 Azure Blob Storage 下载文件。

参数:

local_path (str) – 本地路径。如果下载单个文件，将写入此特定文件，除非它是一个现有目录，此时会在其中创建一个文件。如果下载多个文件，这是写入时的根目录。将根据需要创建目录。
remote_path (str) – 用于查找远程文件的远程路径/globstring。不支持使用 ** 的递归 glob 模式。
nthreads (int) – 使用的线程数。如果为 None，则使用 CPU 核数。
overwrite (bool) – 是否强制覆盖现有文件/目录。如果为 False 且远程路径为目录，则无论是否有文件会被覆盖都会退出。如果为 True，则只覆盖匹配的文件名。
buffersize (int) – int [2**22] 内部缓冲区的字节数。此块不能大于一个 chunk，也不能小于一个 block。
blocksize (int) – int [2**22] 一个块的字节数。在每个 chunk 中，我们为每个 API 调用写入一个较小的块。此块不能大于一个 chunk。

list(path)[source]¶

列出 Azure Data Lake Storage 中的文件。

参数:: path (str) – 用于列出 ADLS 中文件的完整路径/globstring

remove(path, recursive=False, ignore_not_found=True)[source]¶

移除 Azure Data Lake Storage 中的文件。

参数:

path (str) – 要在 ADLS 中移除的目录或文件
recursive (bool) – 是否循环进入该位置的目录并移除文件
ignore_not_found (bool) – 如果找不到要删除的文件是否引发错误

class airflow.providers.microsoft.azure.hooks.data_lake.AzureDataLakeStorageV2Hook(adls_conn_id, public_read=False)[source]¶

基类: airflow.hooks.base.BaseHook

与 ADLS gen2 存储账户交互。

它主要用于在具有分层命名空间的存储账户中创建和管理目录和文件。使用 Adls_v2 连接详情创建 DataLakeServiceClient 对象。

由于 Wasb 被标记为旧版且 ADLS1 已退役，因此最好实现 ADLS gen2 hook 用于与存储账户交互。

另请参阅

https://learn.microsoft.com/en-us/azure/storage/blobs/data-lake-storage-directory-file-acl-python

参数:

adls_conn_id (str) – 引用 adls 连接。
public_read (bool) – 是否应使用匿名公共读取访问。默认为 False

conn_name_attr = 'adls_conn_id'[source]¶

default_conn_name = 'adls_default'[source]¶

conn_type = 'adls'[source]¶

hook_name = 'Azure Date Lake Storage V2'[source]¶

classmethod get_connection_form_widgets()[source]¶

返回要添加到连接表单的连接小部件。

classmethod get_ui_field_behaviour()[source]¶

返回自定义字段行为。

conn_id[source]¶

public_read = False[source]¶

property service_client: azure.storage.filedatalake.DataLakeServiceClient[source]¶

返回 DataLakeServiceClient 对象（已缓存）。

get_conn()[source]¶

返回 DataLakeServiceClient 对象。

create_file_system(file_system_name)[source]¶

在指定账户下创建新的文件系统。

容器充当文件系统。

如果同名文件系统已存在，将引发 ResourceExistsError。此方法返回一个用于与新创建的文件系统交互的客户端。

get_file_system(file_system)[source]¶

获取一个用于与指定文件系统交互的客户端。

参数:: file_system (azure.storage.filedatalake.FileSystemProperties | str) – 这可以是文件系统的名称或 FileSystemProperties 的实例。

create_directory(file_system_name, directory_name, **kwargs)[source]¶

在指定文件系统下创建目录。

参数:

file_system_name (azure.storage.filedatalake.FileSystemProperties | str) – 文件系统的名称或 FileSystemProperties 的实例。
directory_name (str) – 需要在文件系统中创建的目录名称。

get_directory_client(file_system_name, directory_name)[source]¶

获取指定文件系统下的特定目录。

参数:

file_system_name (azure.storage.filedatalake.FileSystemProperties | str) – 文件系统的名称或 FileSystemProperties 的实例。
directory_name (azure.storage.filedatalake.DirectoryProperties | str) – 需要从文件系统中检索的目录名称或 DirectoryProperties 的实例。

create_file(file_system_name, file_name)[source]¶

在文件系统下创建文件。

参数:

file_system_name (azure.storage.filedatalake.FileSystemProperties | str) – 文件系统的名称或 FileSystemProperties 的实例。
file_name (str) – 需要在文件系统中创建的文件名称。

upload_file(file_system_name, file_name, file_path, overwrite=False, **kwargs)[source]¶

在文件系统下创建文件并上传数据。

参数:

file_system_name (azure.storage.filedatalake.FileSystemProperties | str) – 文件系统的名称或 FileSystemProperties 的实例。
file_name (str) – 要创建的文件名称。
file_path (str) – 要加载文件的路径。
overwrite (bool) – 是否覆盖现有文件的布尔标志。

upload_file_to_directory(file_system_name, directory_name, file_name, file_path, overwrite=False, **kwargs)[source]¶

将数据上传到文件。

参数:

file_system_name (str) – 文件系统的名称或 FileSystemProperties 的实例。
directory_name (str) – 目录名称。
file_name (str) – 要创建的文件名称。
file_path (str) – 要加载文件的路径。
overwrite (bool) – 是否覆盖现有文件的布尔标志。

list_files_directory(file_system_name, directory_name)[source]¶

列出指定文件系统下的文件或目录。

参数:

file_system_name (azure.storage.filedatalake.FileSystemProperties | str) – 文件系统的名称或 FileSystemProperties 的实例。
directory_name (str) – 目录名称。

list_file_system(prefix=None, include_metadata=False, **kwargs)[source]¶

列出指定账户下的文件系统。

参数:

prefix (str | None) – 筛选结果，只返回名称以指定前缀开头的文件系统。
include_metadata (bool) – 指定在响应中返回文件系统元数据。默认值为 False。

delete_file_system(file_system_name)[source]¶

删除文件系统。

参数:: file_system_name (azure.storage.filedatalake.FileSystemProperties | str) – 文件系统的名称或 FileSystemProperties 的实例。

delete_directory(file_system_name, directory_name)[source]¶

删除文件系统中指定的目录。

参数:

file_system_name (azure.storage.filedatalake.FileSystemProperties | str) – 文件系统的名称或 FileSystemProperties 的实例。
directory_name (str) – 目录名称。

test_connection()[source]¶: 测试 ADLS Gen2 Storage 连接。