使用 Azure 数据工厂仅复制特定的扩展 blob答案

【问题标题】：Copy only specific extension blobs using Azure Data Factory使用 Azure 数据工厂仅复制特定的扩展 blob
【发布时间】：2017-11-02 12:42:18
【问题描述】：

我正在尝试将 Blob 容器从一个 Azure 存储帐户复制到另一个帐户。我正在使用 Azure Data Factory Copy Activity 来执行此操作。复制所有 blob 很简单。但我只想复制特定的扩展 blob。

在创建输入数据集时，我没有看到任何指定 wildcard 或 regex 的选项。

我有什么方法可以用 ADF 实现这一点。我还尝试了 Azure 数据移动库。即使它没有这样的功能。 DML 中只有基于 prefix 的过滤可用。

【问题讨论】：

标签： azure azure-data-factory azure-blob-storage

【解决方案1】：

在数据集定义中使用 FileFilter 属性来处理这个问题。例如。

{
  "name": "Dataset01",
  "properties": {
    "type": "AzureBlob",
    "linkedServiceName": "BlobStore01",
    "structure": [ ],
    "typeProperties": {
      "folderPath": "FilesFolder1/FilesFolder2",
      "fileFilter": "*.csv" // <<<<< here
      }
      //etc...
     }
     //etc...
    }

这接受使用 * 和 ? 的多字符和单字符通配符。

作为此文档页面的一部分的更多信息：

https://docs.microsoft.com/en-us/azure/data-factory/data-factory-onprem-file-system-connector

希望这会有所帮助。

【讨论】：

很高兴知道存在这样的过滤器。但是你分享的文章提到了Note that fileFilter is applicable for an input FileShare dataset。不确定这是否适用于 Blob 输入数据集。
爆炸！抱歉错过了。我认为如此简单的事情将适用于两者！是时候为 Microsoft 创建另一个用户反馈帖子了 :-)
是的。 ADF 缺少很多基本功能。像带有Last Modified 列的增量备份和按 Blob 类型过滤 Blob。 Azure DML 是目前 Blob 存储的最佳解决方案。

【解决方案2】：

根据我的理解，blob 没有文件扩展名的概念。当我们在本地计算机上下载这些文件时，文件扩展名会很方便。根据文件扩展名，我们的本地计算机可以决定使用哪个应用程序来打开这些文件。

有什么方法可以用 ADF 实现这一点

我们可以使用 Azure 数据工厂自定义活动来做到这一点。我们可以自己实现我们的逻辑。有关如何使用自定义活动的更多信息，请参阅tutorials。

我们也可以使用带有时间触发器的 Azure WebJob 来做到这一点。

如果 Azure 数据工厂是唯一的选择，我们可以使用我们的逻辑来实现复制 blob。以下是我的演示代码。我在我这边测试过，它工作正常

         CloudStorageAccount storageAccountSource = CloudStorageAccount.Parse("connection string");
         CloudStorageAccount storageAccountDest = CloudStorageAccount.Parse("connection string");
         // Create the blob client.
         CloudBlobClient blobClientSource = storageAccountSource.CreateCloudBlobClient();
         CloudBlobClient blobClientDest = storageAccountDest.CreateCloudBlobClient();
         CloudBlobContainer containerSource = blobClientSource.GetContainerReference("test");
         CloudBlobContainer containerDest = blobClientDest.GetContainerReference("test");
         containerDest.CreateIfNotExists();

        SharedAccessBlobPolicy sharedPolicy = new SharedAccessBlobPolicy()
        {

            SharedAccessExpiryTime = DateTime.UtcNow.AddHours(24),
            Permissions = SharedAccessBlobPermissions.Read | SharedAccessBlobPermissions.List |
            SharedAccessBlobPermissions.Write | SharedAccessBlobPermissions.Create | SharedAccessBlobPermissions.Delete
        };

        // Get the container's existing permissions.
        BlobContainerPermissions permissions = containerSource.GetPermissions();
        permissions.SharedAccessPolicies.Add("policy", sharedPolicy);
        containerSource.SetPermissionsAsync(permissions);
        var blobToken = containerSource.GetSharedAccessSignature(sharedPolicy);

        foreach (IListBlobItem item in containerSource.ListBlobs())
        {
            CloudBlob destBlob;
            CloudBlob srcBlob;
            if (item.GetType() == typeof(CloudBlockBlob))
            {
                srcBlob = (CloudBlockBlob)item;
                destBlob = containerDest.GetBlockBlobReference(srcBlob.Name);

            }
            else
            {
                srcBlob = (CloudPageBlob)item;
                destBlob = containerDest.GetPageBlobReference(srcBlob.Name);

            }
            if (srcBlob.Name.Contains("format"))
            {
                destBlob.StartCopy(new Uri(srcBlob.Uri.AbsoluteUri + blobToken));
            }
       }

【讨论】：