S3 效率覆盖与读取

Question

我刚刚完成了以下将客户数据从我的 shopify 获取到 S3 存储桶的功能。现在发生的事情如下。触发器每天运行此 lambda。然后，将所有客户写入 S3 存储桶。每个已经存在的条目都会被覆盖。添加了新客户。

我的问题是：这是一种可扩展的方法，还是我应该读取所​​有文件并比较时间戳以仅添加新条目？还是第二种方法可能更糟？

import requests
import json
import boto3

s3 = boto3.client('s3')
bucket ='testbucket'

url2 = "something.json"

def getCustomers():
    r = requests.get(url2)
    return r.json()

def lambda_handler(event, context):
    
    data = getCustomers()
    
    for customer in data["customers"]:
        
        #create a unique id for each customer
        customer_id = str(customer["id"])
        #create a file name to put the customer in bucket
        file_name = 'customers' + '/' + customer_id + '.json'
        
        #Saving .json to s3
        customer_string = str(customer)
        uploadByteStream = bytes(customer_string.encode('UTF-8')) 
        
        s3.put_object(Bucket=bucket, Key=file_name, Body=uploadByteStream)
        
        
    return {
            'statusCode': 200,
            'body': json.dumps('Success')
        }    

示例响应如下：

{
  "id": 71806090000,
  "email": "something@gmail.com",
  "accepts_marketing": false,
  "created_at": "2021-07-27T11:06:38+02:00",
  "updated_at": "2021-07-27T11:11:58+02:00",
  "first_name": "Bertje",
  "last_name": "Bertens",
  "orders_count": 0,
  "state": "disabled",
  "total_spent": "0.00",
  "last_order_id": null,
  "note": "",
  "verified_email": true,
  "multipass_identifier": null,
  "tax_exempt": false,
  "phone": "+32470000000",
  "tags": "",
  "last_order_name": null,
  "currency": "EUR",
  "addresses": [
    {
      "id": 6623179276486,
      "customer_id": 5371846099142,
      "first_name": "Bertje",
      "last_name": "Bertens",
      "company": "",
      "address1": "Somewhere",
      "address2": "",
      "city": "Somecity",
      "province": null,
      "country": "",
      "zip": "0000",
      "phone": null,
      "name": "Bertje Bertens",
      "province_code": null,
      "country_code": null,
      "country_name": "",
      "default": true
    }
  ],
  "accepts_marketing_updated_at": "2021-07-27T11:11:35+02:00",
  "marketing_opt_in_level": null,
  "tax_exemptions": [],
  "admin_graphql_api_id": "",
  "default_address": {
    "id": 6623179276486,
    "customer_id": 5371846099142,
    "first_name": "Bertje",
    "last_name": "Bertens",
    "company": "",
    "address1": "Somewhere",
    "address2": "",
    "city": "Somecity",
    "province": null,
    "country": "",
    "zip": "0000",
    "phone": null,
    "name": "Bertje Bertens",
    "province_code": null,
    "country_code": null,
    "country_name": "",
    "default": true
  }
}

Answer 1

这是一种可扩展的方法，还是我应该读取所​​有文件并比较时间戳以仅添加新条目？还是第二种方法可能更糟？

一般而言，您不会在使用 Lambda 和 S3 的日常任务中遇到很多可扩展性问题。

一些注意事项：

1. 成本
   一种。 Lambda 执行成本。您的 lambda 运行时间越长，您支付的时间就越多
   湾。 S3 传输成本。除非您在 VPC 中运行 lambda 并为您的存储桶设置 VPC 终端节点，否则您需要从 lambda -> internet (-> s3) 支付 S3 传输费用。

2. Lambda 执行超时。
   如果您有许多文件要上传，您最终可能会遇到一个问题，即您要传输的文件太多，无法在一次调用中完成。

3. 容错
   现在，如果您的 lambda 由于某种原因失败，您将放弃当天的所有工作。

这两种方法对这些考虑有何影响？

对于 (1)，您只需计算成本。从技术上讲，首先检查时间戳的方法将在这里为您提供帮助。但是，我的猜测是，如果您每天只在一次调用中运行它，那么现在的成本是最低的，而且不需要太多关注。我们说的是每月最多几美分（~0.05 美元/月 @ 每天一次完整的 15 分钟调用 + 转移费用）。

对于 (2)，检查时间戳的方法也更好一些，但并不能真正解决可伸缩性问题。如果您预计最终可能会在 Lambda 中耗尽执行时间，您可能需要考虑为解决方案使用新架构。

对于 (3) 这两种方法都没有任何实际意义。无论哪种方式，您都有相同的容错问题。

解决这些领域的可能替代架构组件可能包括：

• 使用 SQS 对文件传输进行排队（有助于解耦和 DLQ 以实现容错）
• 使用计划的（fargate）ECS 任务而不是 Lambda 进行计算（处理 Lambda 超时限制）或让 lambda 批量消耗队列
• S3 VPC 端点和 in-vpc 计算（优化 s3 传输；在更大规模之前可能不具有成本效益）

所以，直接总结一下这个问题：

当前的解决方案存在一些可扩展性问题，即 lambda 的执行超时和容错问题。第二种方法确实引入了优化，但它们没有解决可伸缩性问题。此外，您从第二种解决方案中获得的价值可能并不重要。

无论如何，您提出的建议是有道理的，不应该花费太多精力来实施。

...
customer_updated_at = datetime.datetime.fromisoformat(customer['created_at'])

file_name = 'customers' + '/' + customer_id + '.json'

# Send HEAD request to check date to see if we need to update it
response = s3.head_object(bucket, file_name)
s3_modified = response["LastModified"]
if customer_updated_at > s3_modified:
    # Saving .json to s3
    customer_string = str(customer)
    uploadByteStream = bytes(customer_string.encode('UTF-8'))
    s3.put_object(Bucket=bucket, Key=file_name, Body=uploadByteStream)
else:
    print('s3 version is up to date, no need to upload')

Answer 2

只要您设法在 Lambda 的最长 15 分钟超时时间内完成整个过程，它就会起作用。 S3 旨在扩展到要求更高的工作负载；-)

但是：

正如您已经观察到的那样，它的效率非常低。更好的实现是在某处跟踪最后一次完全加载的时间戳，例如DynamoDB 或 Systems Manager 参数存储，并且仅写入 "created_at" 或 "updated_at" 属性在最后一次成功完全加载之后的所有客户。最后你更新完整的加载时间戳。

这是一些伪代码：

last_full_load_date = get_last_full_load() or '1900-01-01T00:00:00Z'

customers = get_customers()

for customer in customers:
    if customer.created_at >= last_full_load_date or customer.updated_at >= last_full_load_date:
        write_customer(customer)

set_last_full_load(datetime.now())

这样你只写实际改变的数据（假设 API 是可靠的）。

这还有一个好处，即如果在写入过程中出现问题，您可以重试，因为您最后只更新了 last_full_load 时间。或者，您可以跟踪每个用户的最后修改时间，但如果您要批量加载，这似乎没有必要。