使用PySpark写入DynamoDB时避免AttributeValues-Python教程-PHP中文网

使用pyspark写入dynamodb时避免attributevalues

本文介绍了如何在使用PySpark将数据写入DynamoDB时，避免生成DynamoDB-JSON格式的数据，即去除AttributeValues。核心在于理解DynamoDB的数据存储格式，以及如何通过数据转换或使用合适的SDK来达到所需的结果，最终实现将数据以更简洁的JSON格式写入DynamoDB。

在使用PySpark将数据写入DynamoDB时，默认情况下，数据会以DynamoDB-JSON格式存储。这种格式包含了类型描述符，例如{ "S" : "string_value" }表示字符串类型，{ "N" : "123" }表示数字类型。然而，有时我们希望以更简洁的JSON格式存储数据，例如直接存储"string_value"或123，而不需要类型描述符。以下是如何实现这一目标的方法：

理解DynamoDB数据模型

首先，需要理解DynamoDB存储数据的底层模型。DynamoDB始终使用DynamoDB-JSON格式存储数据。这种格式是为了让DynamoDB能够明确区分不同数据类型，并进行高效的存储和检索。

问题分析：为什么会出现AttributeValues？

当你使用AWS Glue的write_dynamic_frame_from_options方法将PySpark DataFrame写入DynamoDB时，Glue会自动将数据转换为DynamoDB-JSON格式。这是因为Glue的设计目标是处理各种数据源，并将其转换为DynamoDB能够理解的格式。

解决方案：数据转换和SDK选择

要避免AttributeValues，主要有两种方法：

数据转换： 在写入DynamoDB之前，将数据转换为所需的JSON格式。
使用合适的SDK： 选择能够直接写入所需格式的SDK。

方法一：数据转换

jm-jsjkxyjs02-zlm-405

使用约定式提交（Conventional Commits）生成标准化的 Git 提交信息。适用于以下场景： (1) 暂存更改时需要提交信息； (2) 审查 diff 时希望获得结构化摘要； (3) 编写 PR 描述或变更日志（changelog）条目； (4) 用户明确要求“为本次更改编写提交信息”。该功能会分析已暂存/未暂存的 diff，并生成按类型（type）划分范围的提交信息。

下载

在写入DynamoDB之前，可以使用PySpark的转换函数将数据转换为所需的格式。以下是一个示例，展示如何将数组中的字符串转换为普通字符串数组：

from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType, StringType

def remove_attribute_values(data):
  """
  移除DynamoDB AttributeValues.
  """
  if isinstance(data, list):
    return [remove_attribute_values(item) for item in data]
  elif isinstance(data, dict):
    if "S" in data:
      return data["S"]
    elif "N" in data:
      return data["N"]
    elif "L" in data:
      return remove_attribute_values(data["L"])
    else:
      return data
  else:
    return data

remove_attribute_values_udf = udf(remove_attribute_values, ArrayType(StringType()))

# 假设 df 是你的 DataFrame, 'data3' 是包含数组的列
df = df.withColumn("data3_transformed", remove_attribute_values_udf(df["data3"]))

# 现在使用 data3_transformed 列写入 DynamoDB
glue_context.write_dynamic_frame_from_options(
    frame=DynamicFrame.fromDF(df.drop("data3"), glue_context, "output"), # 移除原始的 data3 列
    connection_type="dynamodb",
    connection_options={
        "dynamodb.output.tableName": "table_name",
        "dynamodb.throughput.write.percent": "1.0",
    },
)

注意： 上述代码示例需要根据你的具体数据结构进行调整。你需要确保remove_attribute_values函数能够正确处理你的数据类型。

方法二：使用合适的SDK

另一种方法是使用能够直接写入所需格式的SDK。例如，可以使用boto3库直接与DynamoDB交互。

import boto3
import json

dynamodb = boto3.resource('dynamodb', region_name='your_region') # 替换为你的区域
table = dynamodb.Table('table_name') # 替换为你的表名

def write_to_dynamodb(data):
    """
    使用boto3写入DynamoDB，不使用AttributeValues。
    """
    table.put_item(Item=data)

# 假设 df 是你的 DataFrame
for row in df.collect():
    data = row.asDict()
    # 可以选择性地对data进行转换，例如将array类型转换为list
    write_to_dynamodb(data)

注意： 使用boto3时，你需要自己处理数据的序列化和写入过程。这需要你对DynamoDB的API有更深入的了解。

总结和注意事项