从S3批量导入Parquet数据到DynamoDB的成本优化问询

更新于02/06 #

目前已完成Parquet转压缩CSV并上传至S3的步骤，代码如下：

from pyspark.sql import functions as F

import zlib
import json
import boto3
from datetime import datetime

from pyspark import SparkContext, SparkConf
from pyspark.sql import SQLContext
import pyspark.sql.functions as F
from pyspark import sql
import os

# 调用外部函数获取AWS凭证
s3_client = get_boto(spark)

small_df = spark.read.format('parquet').options(header=True,inferSchema=True).load('path-to-s3')

# 将所有列转换为String类型
to_string_df = small_df.select([F.col(c).cast("string") for c in small_df.columns])
# 上传至另一个S3存储桶为压缩CSV文件
to_string_df.coalesce(1).write.mode("overwrite").option("compression","gzip").csv('<path-to-s3>')

将压缩CSV上传至S3后，手动触发向DynamoDB导入并创建新表时，遇到以下错误：

{
    "itemS3Pointer": {
        "bucket": "my-s3-bucket",
        "key": "csv-file-name.csv",
        "itemIndex": 9059
    },
    "importArn": "arn:aws:dynamodb:****",
    "errorMessages": [
        "Item size has exceeded the maximum allowed size"
    ]
}

后续会持续更新解决方案。

场景说明 #

• 单个S3存储桶包含300+对象，总大小1GB-2.5GB，后续会有多个同类存储桶，当前以单个为例。

核心需求 #

1. 从S3批量导入数据到DynamoDB，参考过DynamoDB官方的S3数据导入方案，但该方案不支持Parquet格式
2. 理想目标：每次导入时创建新表，从S3批量加载数据到新表

已尝试的方案 #

方案1：Parquet转Pandas再批量写入DynamoDB #

步骤：

• 从S3读取Parquet文件
• 转换为本地Pandas DataFrame
• 将DataFrame行转为JSON字符串
• 使用batch_writer写入DynamoDB
  问题：DynamoDB写入成本过高，无法接受

方案2：Parquet转CSV后用DynamoDB官方导入（当前进行中） #

即上述02/06更新的方案，目前卡在单条数据大小超过DynamoDB限制的问题

备选方案疑问 #

考虑用AWS Glue任务读取S3中的Parquet数据，转换后写入DynamoDB，但不确定该方案是否会消耗WriteCapacity导致成本失控。

恳请社区提供指导：我的思路是否存在问题？有没有更优的解决方案？

内容的提问来源于stack exchange，提问作者ApJo