Скопируйте данные из хранилища BLOB-объектов Azure в AWS S3
Я новичок в фабрике данных Azure и у меня есть интересное требование.
Мне нужно переместить файлы из хранилища BLOB-объектов Azure в Amazon S3, в идеале с помощью фабрики данных Azure.
Однако S3 не поддерживается как приемник;
https://docs.microsoft.com/en-us/azure/data-factory/copy-activity-overview
Я также понимаю из множества комментариев, которые я прочитал здесь, что вы не можете напрямую скопировать из хранилища BLOB-объектов на S3 - вам нужно будет загрузить файл локально, а затем загрузить его на S3.
Кто-нибудь знает какие-либо примеры в фабрике данных, службах SSIS или Azure Runbook, которые могут сделать такую вещь, я полагаю, что можно было бы написать приложение или функцию Azure для логики, которая вызывается из фабрики данных.
4 ответа
Удалось заставить что-то работать над этим - это может пригодиться кому-то еще.
Я решил написать лазурную функцию, которая использует HTTP-запрос в качестве триггера.
Эти два поста мне очень помогли;
Как я могу использовать пакеты NuGet в своих функциях Azure?
Скопируйте из BLOB-объекта Azure в AWS S3 с помощью C#
Обратите внимание на мой ответ на пакеты Nuget, если вы используете функции Azure 2.x.
Вот код - вы можете изменить основу для ваших нужд. Я возвращаю сериализованный объект JSON, потому что фабрика данных Azure требует этого как ответ от http-запроса, отправленного из конвейера;
#r "Microsoft.WindowsAzure.Storage"
#r "Newtonsoft.Json"
#r "System.Net.Http"
using System.Net;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Primitives;
using Newtonsoft.Json;
using Microsoft.WindowsAzure.Storage.Blob;
using System.Net.Http;
using Amazon.S3;
using Amazon.S3.Model;
using Amazon.S3.Transfer;
using Amazon.S3.Util;
public static async Task<IActionResult> Run(HttpRequest req, ILogger log)
{
log.LogInformation("Example Function has recieved a HTTP Request");
// get Params from query string
string blobUri = req.Query["blobUri"];
string bucketName = req.Query["bucketName"];
// Validate query string
if (String.IsNullOrEmpty(blobUri) || String.IsNullOrEmpty(bucketName)) {
Result outcome = new Result("Invalid Parameters Passed to Function",false,"blobUri or bucketName is null or empty");
return new BadRequestObjectResult(outcome.ConvertResultToJson());
}
// cast the blob to its type
Uri blobAbsoluteUri = new Uri(blobUri);
CloudBlockBlob blob = new CloudBlockBlob(blobAbsoluteUri);
// Do the Copy
bool resultBool = await CopyBlob(blob, bucketName, log);
if (resultBool) {
Result outcome = new Result("Copy Completed",true,"Blob: " + blobUri + " Copied to Bucket: " + bucketName);
return (ActionResult)new OkObjectResult(outcome.ConvertResultToJson());
}
else {
Result outcome = new Result("ERROR",false,"Copy was not successful Please review Application Logs");
return new BadRequestObjectResult(outcome.ConvertResultToJson());
}
}
static async Task<bool> CopyBlob(CloudBlockBlob blob, string existingBucket, ILogger log) {
var accessKey = "myAwsKey";
var secretKey = "myAwsSecret";
var keyName = blob.Name;
// Make the client
AmazonS3Client myClient = new AmazonS3Client(accessKey, secretKey, Amazon.RegionEndpoint.EUWest1);
// Check the Target Bucket Exists;
bool bucketExists = await AmazonS3Util.DoesS3BucketExistAsync (myClient,existingBucket);
if (!bucketExists) {
log.LogInformation("Bucket: " + existingBucket + " does not exist or is inaccessible to the application");
return false;
}
// Set up the Transfer Utility
TransferUtility fileTransferUtility = new TransferUtility(myClient);
// Stream the file
try {
log.LogInformation("Starting Copy");
using (var stream = await blob.OpenReadAsync()) {
// Note: You need permissions to not be private on the source blob
log.LogInformation("Streaming");
await fileTransferUtility.UploadAsync(stream,existingBucket,keyName);
log.LogInformation("Streaming Done");
}
log.LogInformation("Copy completed");
}
catch (AmazonS3Exception e) {
log.LogInformation("Error encountered on server. Message:'{0}' when writing an object", e.Message);
}
catch (Exception e) {
log.LogInformation("Unknown encountered on server. Message:'{0}' when writing an object", e.Message);
return false;
}
return true;
}
public class Result {
public string result;
public bool outcome;
public string UTCtime;
public string details;
public Result(string msg, bool outcomeBool, string fullMsg){
result=msg;
UTCtime=DateTime.Now.ToString("yyyy-MM-dd h:mm:ss tt");
outcome=outcomeBool;
details=fullMsg;
}
public string ConvertResultToJson() {
return JsonConvert.SerializeObject(this);
}
}
ADF теперь включает SFTP в качестве приемника. По той же ссылке, указанной в вопросе (в качестве приемника поддерживается крайний правый столбец):
Используя семейство AWS Transfer, вы можете настроить SFTP-сервер и добавить пользователя с открытым ключом SSH, а затем использовать эту конфигурацию для настройки SFTP-соединения из ADF , которое будет напрямую подключаться к корзине S3.
Вы можете использовать Skyplane для копирования данных между облаками (ускорение в 110 раз по сравнению с инструментами CLI, с автоматическим сжатием для экономии при выходе). Чтобы перейти из хранилища BLOB-объектов Azure в S3, вы можете вызвать одну из команд:
skyplane cp -r az://azure-bucket-name/ s3://aws-bucket-name/
skyplane sync -r az://azure-bucket-name/ s3://aws-bucket-name/
Скачивание файлов из хранилища Azure с помощью AzCopy во временный локальный репозиторий
Вы можете загрузить файлы из облачного хранилища Azure в свою локальную систему, просто следуйте приведенной ниже команде, используйте рекурсивный флаг, чтобы скопировать все файлы.
azcopy /Source:[source_container_url] /Dest:[local_file_path] /Sourcekey:[source_storage_account_access_key] /s
Загрузите локальные файлы в Amazon S3 с помощью команды aws s3 cp
aws s3 cp local_file_path s3://my-bucket/ --recursive