PutDynamoDBRecord

说明

PutDynamoDBRecord 提供了从面向记录的 FlowFile 向 DynamoDB 表插入多个 Item 的能力。与 PutDynamoDB 相比，该 Processor 能够处理 JSON 之外其他 格式的数据，并能为给定 Item 添加多个字段。此外，PutDynamoDBRecord 设计上支持向数据库插入更大批次的数据。

数据类型

DynamoDB 支持的数据类型列表与 Record 数据结构的能力并不完全重合。插入 数据时需要进行一些转换和简化，包括：

• 数值在 Item 内以浮点数据结构存储。在某些情况下，这种表示可能带来 精度问题。
• Char 不是 DynamoDB 支持的类型，这些字段会被转换为 String 值。
• Enum 类型存储为 String 字段，使用枚举的名称。
• DynamoDB 把时间和日期相关信息存储为 String。
• 内部 record 结构会被转换为 map。
• Choice 不是受支持的数据类型，无论实际包裹的数据类型是什么，包裹在 Choice 中的值都会按 String 处理。
• 未知数据类型按字符串处理。

限制

使用 DynamoDB 批量插入有两个固有限制。首先，任何情况下一次插入的 Item 数量都限制为 25。为了克服这一点，PutDynamoDBRecord 在一次 执行中会根据传入 FlowFile 的记录数量，向数据库服务器发起多次插入 调用。采用这种方式，流程在大多数情况下不必关心这个限制。

多次外部操作伴随着某一步出现意外结果的风险。例如，传入 FlowFile 包含 70 条记录时，会被拆成 3 块，每块执行一次插入操作。前两块每块 包含 25 个待插入 Item，第三块包含剩余的 20 个。某些情况下可能出现 前两次插入成功而第三次失败。此时我们把该 FlowFile 视为"部分处理"， 并根据问题的性质把它转移到 "failure" 或 "unprocessed" Relationship。 为了保留已成功处理块的信息，Processor 会给 FlowFile 赋 "dynamodb.chunks.processed" Attribute，值为成功处理的块数。

该行为最常见的原因来自 DynamoDB 插入的另一个限制：数据库对插入的 数据量有内置的管控。当客户端达到"吞吐量上限"时，服务器会在一段 时间内拒绝处理插入请求。更多信息参见这里。 从 PutDynamoDBRecord 的角度，我们把这类情况视为临时问题， FlowFile 会被转移到 "unprocessed" Relationship，之后 Processor 会让出（yield）以避免进一步的吞吐量问题。（其他类型的失败会导致 转移到 "failure" Relationship。）

分区键和排序键

该 Processor 支持多种为插入的 Item 分配分区键和排序键的策略：

分区键策略

Partition By Field

Processor 把某个记录字段指定为分区键。记录字段的名称由 "Partition Key Field" 属性指定，值为同名记录字段的值。

Partition By Attribute

Processor 把某个 FlowFile Attribute 的值指定为分区键。使用该策略时， FlowFile 内的所有 Item 共享同一个分区键值。建议配合同时具有排序键的 表使用，以满足 DynamoDB 的主键要求。"Partition Key Field" 属性定义 Item 字段的名称，"Partition Key Attribute" 属性指定哪个 Attribute 的值会被赋给分区键。使用该策略时，"Partition Key Field" 必须与 传入记录包含的字段不同。

Generated UUID

使用该策略时，Processor 会为每个 Item 生成一个 UUID 标识符，并把 它作为分区键的值。用作分区键的字段名由 "Partition Key Field" 属性 定义。使用该策略时，"Partition Key Field" 必须与传入记录包含的 字段不同。使用该策略时，DynamoDB 表中的分区键必须是 String 类型。

排序键策略

None

不给 Item 分配排序键。如果表定义要求排序键，使用该策略会导致插入失败。

Sort By Field

Processor 把某个记录字段指定为排序键。记录字段的名称由 "Sort Key Field" 属性指定，值为同名记录字段的值。使用该策略时， "Sort Key Field" 必须与传入记录包含的字段不同。

Generate Sequence

Processor 根据原始记录在传入 FlowFile 中的位置（与分块无关）为每个 Item 分配一个生成的值。第一个 Item 的排序键为 1，第二个为 2，以此 类推。生成的键在给定 FlowFile 内唯一。记录字段的名称由 "Sort Key Field" 属性指定。使用该策略时，"Sort Key Field" 必须与 传入记录包含的字段不同。使用该策略时，DynamoDB 表中的排序键必须是 Number 类型。

示例

使用字段作为分区键和排序键

配置

• Partition Key Strategy: Partition By Field
• Partition Key Field: class
• Sort Key Strategy: Sort By Field
• Sort Key Field: size

注意：两个字段都必须存在于传入记录中！

结果

使用这一对策略得到的 Item 与传入记录一致（不计转换带来的表示变化）。 属性指定的字段会正常添加到 Item，唯一的区别是被标记为（主）键项。

输入

[
  {
    "type": "A",
    "subtype": 4,
    "class": "t",
    "size": 1
  }
]

输出（示意）

• type: 值为 "A" 的 String 字段
• subtype: 值为 4 的 Number 字段
• class: 值为 "t" 的 String 字段，作为分区键
• size: 值为 1 的 Number 字段，作为排序键

使用 FlowFile 文件名作为分区键，排序键自动生成

配置

• Partition Key Strategy: Partition By Attribute
• Partition Key Field: source
• Partition Key Attribute: filename
• Sort Key Strategy: Generate Sequence
• Sort Key Field: sort

结果

FlowFile 的 filename Attribute 会被用作分区键。此时同一 FlowFile 内的所有记录共享同一个分区键。为避免冲突，如果 FlowFile 包含多条记录，建议使用排序键。此时使用生成的序列， 它在给定 FlowFile 内保证唯一。

输入

[
  {
    "type": "A",
    "subtype": 4,
    "class": "t",
    "size": 1
  },
  {
    "type": "B",
    "subtype": 5,
    "class": "m",
    "size": 2
  }
]

输出（示意）

第一个 Item

• source: 值为 "data46362.json" 的 String 字段，作为分区键
• type: 值为 "A" 的 String 字段
• subtype: 值为 4 的 Number 字段
• class: 值为 "t" 的 String 字段
• size: 值为 1 的 Number 字段
• sort: 值为 1 的 Number 字段，作为排序键

第二个 Item

• source: 值为 "data46362.json" 的 String 字段，作为分区键
• type: 值为 "B" 的 String 字段
• subtype: 值为 5 的 Number 字段
• class: 值为 "m" 的 String 字段
• size: 值为 2 的 Number 字段
• sort: 值为 2 的 Number 字段，作为排序键

使用生成的分区键

配置

• Partition Key Strategy: Generated UUID
• Partition Key Field: identifier
• Sort Key Strategy: None

结果

会生成一个 UUID 作为分区键，每个 Item 生成的 UUID 都不同。

输入

[
  {
    "type": "A",
    "subtype": 4,
    "class": "t",
    "size": 1
  }
]

输出（示意）

• identifier: 值为 "872ab776-ed73-4d37-a04a-807f0297e06e" 的 String 字段，作为分区键
• type: 值为 "A" 的 String 字段
• subtype: 值为 4 的 Number 字段
• class: 值为 "t" 的 String 字段