ListS3

流式处理与批处理

ListS3 会对所配置 S3 桶中遇到的所有 S3 对象执行列表（listing）操作。 常见的用例大致分为两类。

流式用例

默认情况下，该 Processor 会为桶中的每个对象创建一个独立的 FlowFile，并添加 filename、bucket 等 Attribute。一个常见用法是把 ListS3 连接到 FetchS3 Processor。这两个 Processor 配合使用，可以 方便地监视桶，并在任何新对象到达 S3 时以高效的流式方式获取其内容。

批处理用例

另一个常见用例是：希望处理给定桶中所有新到达的对象，并且只有当所有 对象都处理完成之后才执行某些操作。上述流式数据的处理方式使这一点 变得困难，因为 Lynxus 本质上是一个流式平台，不存在有明确起点和终点 的"作业"，数据只是在可用时被拾取。

为解决这个问题，ListS3 Processor 可以可选地配置 Record Writer。 配置后，只会创建单个 FlowFile，其中为桶中的每个对象包含一条记录， 而不是每个对象一个 FlowFile。关于如何构建"先处理完所有对象再进入 下一步"的数据流，可参考 ListFile 的文档示例。

不过，ListFile 与 ListS3 产生的数据有一个重要区别：两者输出的记录 结构不同。ListS3 输出的记录遵循以下 schema（Avro 格式）：

{
  "type": "record",
  "name": "lynxusRecord",
  "namespace": "cn.org.acal.lynxus",
  "fields": [
    {
      "name": "key",
      "type": "string"
    },
    {
      "name": "bucket",
      "type": "string"
    },
    {
      "name": "owner",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "etag",
      "type": "string"
    },
    {
      "name": "lastModified",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    },
    {
      "name": "size",
      "type": "long"
    },
    {
      "name": "storageClass",
      "type": "string"
    },
    {
      "name": "latest",
      "type": "boolean"
    },
    {
      "name": "versionId",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "tags",
      "type": [
        "null",
        {
          "type": "map",
          "values": "string"
        }
      ]
    },
    {
      "name": "userMetadata",
      "type": [
        "null",
        {
          "type": "map",
          "values": "string"
        }
      ]
    }
  ]
}