从 azure blob 并行读取多个文件

Question

我希望从 azure blob 中读取大量小文件，这可能是 1k-100k 个文件的顺序，总共有几个 1TB。我必须在 python 中处理这些文件，它本身的处理并不繁重，但是从 blob 中读取文件确实需要时间。另一个限制是，当我处理第一个文件时，新文件已被写入。

我正在寻找执行此操作的选项，是否可以使用 dask 从 blob 并行读取多个文件？或者是否可以在 azure 网络内每小时传输和加载超过 1tb 的数据？

Answer 1

好吧，您有几个选项可以在这里实现并行性：

多线程：

下面使用ThreadPoolPython 中的类从 Azure 存储并行下载和处理文件。注意：使用v12存储sdk

import os
from multiprocessing.pool import ThreadPool
from azure.storage.blob import BlobServiceClient

STORAGE_CONNECTION_STRING = "REPLACE_THIS"
BLOB_CONTAINER = "myfiles"

class AzureBlobProcessor:
  def __init__(self): 
    # Initialize client
    self.blob_service_client =  BlobServiceClient.from_connection_string(STORAGE_CONNECTION_STRING)
    self.blob_container = self.blob_service_client.get_container_client(BLOB_CONTAINER)
 
  def process_all_blobs_in_container(self):
    # get a list of blobs
    blobs = self.blob_container.list_blobs()
    result = self.execute(blobs)
 
  def execute(self, blobs):
    # Just sample number of threads as 10
    with ThreadPool(processes=int(10)) as pool:
     return pool.map(self.download_and_process_blob, blobs)
 
  def download_and_process_blob(self,blob):
    file_name = blob.name
    
    # below is just sample which reads bytes, update to variant you need
    bytes = self.blob_container.get_blob_client(blob).download_blob().readall()
 
    # processing logic goes here :)

    return file_name
 
# caller code
azure_blob_processor = AzureBlobProcessor()
azure_blob_processor.process_all_blobs_in_container()

您还可以查看dask 远程数据读取。检查https://github.com/dask/adlfs

要使用 Gen1 文件系统：

import dask.dataframe as dd

storage_options={'tenant_id': TENANT_ID, 'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET}

dd.read_csv('adl://{STORE_NAME}/{FOLDER}/*.csv', storage_options=storage_options)

要使用 Gen2 文件系统，您可以使用协议abfs或az：

import dask.dataframe as dd

storage_options={'account_name': ACCOUNT_NAME, 'account_key': ACCOUNT_KEY}

ddf = dd.read_csv('abfs://{CONTAINER}/{FOLDER}/*.csv', storage_options=storage_options)
ddf = dd.read_parquet('az://{CONTAINER}/folder.parquet', storage_options=storage_options)

要从公共存储 blob 读取，您需要指定'account_name'. 例如，您可以通过以下方式访问纽约市出租车和豪华轿车委员会：

storage_options = {'account_name': 'azureopendatastorage'}
ddf = dd.read_parquet('az://nyctlc/green/puYear=2019/puMonth=*/*.parquet', storage_options=storage_options)

利用 Azure PaaS 实现并行性：

嗯，在这条路上你有多种选择。

• Azure Batch：教程：使用 Python API 通过 Azure Batch 运行并行工作负载
• 具有 Blob 事件网格触发器的 Azure Function：Azure Functions 的 Azure 事件网格绑定

最后，我建议您深入研究Blob 存储的性能和可扩展性清单，以确保您在 Azure 存储帐户数据传输的限制范围内。还有标准存储帐户的可扩展性和性能目标。查看您每小时 1 TB 的要求，如果您从上述文档转换为 GBPS，则似乎处于限制之下。