3. 对象存储跨云复制实战：AWS S3 → Azure Blob

跨云复制这件事，说白了就是让两个云厂商的存储桶“对齐”。我见过不少团队，数据在 S3 上跑得好好的，突然要迁一部分到 Azure Blob，原因嘛——要么是成本，要么是合规。今天我们就拿 AWS S3 到 Azure Blob 这条线，走一遍实战。

3.1 整体架构

先看一张图，把流程理清楚。我个人习惯在动手前先画架构，不然代码写到一半容易跑偏。

流程其实不复杂：S3 桶里一有对象上传或修改，事件通知就发给 Lambda。Lambda 拿到事件后，用 Azure Blob SDK 把数据写到 Azure 那边。嗯，这里要注意——Lambda 的执行环境需要能访问公网，或者通过 NAT 网关出去。

3.2 前置准备

动手之前，先把两边的东西准备好。我列个清单，你照着配就行。

平台	需要准备的内容	说明
AWS	S3 源桶 + 事件通知配置	开启事件到 SQS / Lambda
AWS	Lambda 函数 + IAM 角色	权限：S3 读取 + 日志写入
Azure	存储账户 + Blob 容器	记录连接字符串或 SAS
Azure	服务主体（可选）	用于托管身份认证

3.3 核心代码实现

代码部分，我用 Python 写 Lambda 函数。你想想看，Lambda 本身就是事件驱动的，跟 S3 通知是天生一对。

import boto3
import os
from azure.storage.blob import BlobServiceClient
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3 = boto3.client('s3')

# Azure 连接配置
AZURE_CONN_STRING = os.environ['AZURE_STORAGE_CONNECTION_STRING']
AZURE_CONTAINER = os.environ['AZURE_TARGET_CONTAINER']

def lambda_handler(event, context):
    # 解析 S3 事件记录
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        
        logger.info(f"收到同步请求: s3://{bucket}/{key}")
        
        # 从 S3 读取对象
        try:
            response = s3.get_object(Bucket=bucket, Key=key)
            data = response['Body'].read()
            metadata = response.get('Metadata', {})
        except Exception as e:
            logger.error(f"读取 S3 对象失败: {str(e)}")
            continue
        
        # 上传到 Azure Blob
        try:
            blob_service = BlobServiceClient.from_connection_string(AZURE_CONN_STRING)
            blob_client = blob_service.get_blob_client(
                container=AZURE_CONTAINER, 
                blob=key
            )
            
            # 上传并设置元数据
            blob_client.upload_blob(data, overwrite=True)
            blob_client.set_blob_metadata(metadata)
            
            logger.info(f"同步成功: {key} → Azure Blob")
        except Exception as e:
            logger.error(f"上传到 Azure 失败: {str(e)}")
            continue
    
    return {'statusCode': 200, 'body': '同步完成'}

这段代码我解释一下关键点：

• 事件解析：S3 事件通知的格式是固定的，Record 数组里每个元素包含桶名和对象键。
• 元数据传递：我把 S3 对象的自定义元数据原样带到 Azure Blob，这样两边保持一致。
• 覆盖写入：overwrite=True 确保同名文件更新时不会报错。

3.4 大文件分块上传

超过 100MB 的文件，直接 upload_blob 容易内存溢出。我建议用分块上传。说白了就是把文件切成小块，一块一块传，失败了还能重试。

from azure.storage.blob import BlobClient
import math

def upload_large_file(blob_client, file_stream, file_size):
    block_size = 4 * 1024 * 1024  # 4MB 一块
    block_count = math.ceil(file_size / block_size)
    block_ids = []
    
    for i in range(block_count):
        start = i * block_size
        end = min(start + block_size, file_size)
        data = file_stream.read(end - start)
        
        block_id = f"block-{i:06d}"
        blob_client.stage_block(block_id, data)
        block_ids.append(block_id)
        
        logger.info(f"分块 {i+1}/{block_count} 上传完成")
    
    # 提交块列表
    blob_client.commit_block_list(block_ids)
    logger.info("所有分块提交完成")

分块上传的好处很明显：

• 内存占用可控，不会因为一个大文件把 Lambda 搞崩。
• 网络中断后只重传失败的块，不用从头再来。
• 可以并行上传多个块，速度更快。

3.5 增量同步与去重

全量同步一次就够了，后面都是增量。怎么判断文件变了？我一般用 ETag 或 LastModified 做对比。

对比方式	优点	缺点
ETag	精确，文件内容变了 ETag 就变	分块上传的 ETag 不是 MD5
LastModified	简单，直接比较时间戳	时间精度到秒，可能漏掉毫秒级变更
自定义元数据	灵活，可以记录同步版本	需要额外维护

我个人习惯用 LastModified 做第一道过滤，再配合 ETag 做精确校验。代码里可以这样写：

def should_sync(s3_obj, blob_client):
    """判断是否需要同步"""
    # 获取 S3 对象信息
    s3_last_modified = s3_obj['LastModified']
    s3_etag = s3_obj['ETag']
    
    try:
        # 获取 Azure Blob 属性
        blob_properties = blob_client.get_blob_properties()
        blob_last_modified = blob_properties.last_modified
        blob_etag = blob_properties.etag
        
        # 比较
        if s3_last_modified <= blob_last_modified and s3_etag == blob_etag:
            return False
    except Exception:
        # Blob 不存在，需要同步
        pass
    
    return True

3.6 错误处理与重试机制

跨云同步最怕网络抖动。Azure 那边偶尔会返回 503，S3 读取也可能超时。我的做法是加指数退避重试。

import time
import random

def retry_with_backoff(func, max_retries=3):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            logger.warning(f"重试 {attempt+1}/{max_retries}，等待 {wait:.2f} 秒")
            time.sleep(wait)

用法很简单：把上传操作包在 retry_with_backoff 里就行。第一次失败等 2 秒，第二次等 4 秒，第三次等 8 秒。三次都失败就记日志，人工介入。

3.7 成本优化要点

跨云复制不是免费的。我算过一笔账，给你参考：

成本项	AWS 侧	Azure 侧
S3 GET 请求	每 1000 次约 $0.0004	—
S3 数据传输出	每 GB 约 $0.09（到互联网）	—
Lambda 执行	每 100 万次约 $0.20	—
Azure Blob 写入	—	每 10000 次约 $0.05
Azure 存储	—	每 GB/月约 $0.018

你看，大头其实是 S3 的数据传出费。如果每天同步几百 GB，一个月下来几千美金就没了。怎么省？

• 只同步增量：别全量跑，用上面说的去重逻辑。
• 压缩再传：文本类文件先 gzip，能省 60%-80% 流量。
• 选对区域：如果 AWS 和 Azure 都有香港区域，走内网对等连接，传出费几乎为零。