ListFTP

ListFTP 会对 FTP 服务器上配置目录中遇到的所有文件执行列表（listing）操作。常见的用例大致可分为两类。

流式用例

默认情况下，该 Processor 会为目录中的每个文件创建一个独立的 FlowFile，并添加 filename、path 等 Attribute。一个常见用法是把 ListFTP 连接到 FetchFTP Processor。这两个 Processor 配合使用， 可以方便地监视目录，并在任何新文件到达 FTP 服务器时以高效的流式方式获取其内容。

批处理用例

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

为解决这一问题，ListFTP Processor 可以可选地配置一个 Record Writer。配置 Record Writer 后， 只会创建单个 FlowFile，其中为目录中的每个文件包含一条 Record，而不是每个文件一个 FlowFile。 在这种模式下，要获取每个文件的内容，必须先把记录拆分为独立的 FlowFile，然后再逐个获取。 那么这样做有什么帮助呢？

我们仍然可以实现"等待目录中所有文件处理完毕"的目标：把 FlowFile 拆开，并在一个 Process Group 中处理所有数据。将 Process Group 的 FlowFile Concurrency 配置为 "Single FlowFile per Node"， 意味着一次只会有一个 FlowFile 进入该 Process Group。进入之后，这个 FlowFile 可以被拆分并逐部分 处理。再将 Process Group 的 Outbound Policy 配置为 "Batch Output"，意味着在所有数据处理完成 之前，任何 FlowFile 都不会离开该 Process Group。

在这个流程中，我们用 ListFTP 对目录做列表操作。该 Processor 配置了一个 Record Writer（本例 使用 CSV Writer，但任何 Record Writer 都可以），这样整个列表只会生成一个 FlowFile。这份列表 随后被发送到 "Process Listing" Process Group（如下所示）。只有当整个目录的内容都处理完毕后， 数据才会离开 "Process Listing" Process Group。此时，当 Process Group 中的所有数据都已准备 离开时，每个处理完成的文件会被发送到 "Post-Processing" Process Group；同时，原始的列表会被 发送到 "Processing Complete Notification" Process Group。要实现这一点，Process Group 必须 配置 FlowFile Concurrency 为 "Single FlowFile per Node"、Outbound Policy 为 "Batch Output"。

在 "Process Listing" Process Group 中，列表通过 "Listing" Input Port 接收，然后直接发送到 "Listing of Processed Data" Output Port，以便在所有处理完成时，原始列表也一并输出。

接下来，列表被拆分为每条记录一个独立的 FlowFile。因为我们要用 FetchFTP 获取数据，所以需要把 文件的 filename 和 path 作为 FlowFile Attribute。实现方式有几种，最简单的机制是使用 PartitionRecord Processor。该 Processor 配置了一个能够读取 ListFTP 所写数据的 Record Reader （本例为 CSV Reader），并配置了两个额外的用户自定义属性：

• path: /path
• filename: /filename

这样，进入 PartitionRecord Processor 的每条记录都会被拆分为一个独立的 FlowFile（因为 "path" 与 "filename" 字段的组合对每条记录都是唯一的），"filename" 和 "path" 记录字段会成为 FlowFile 上的 Attribute。FetchFTP 的 "Remote File" 属性配置为使用 ${path}/${filename}， 从而利用这些 Attribute。

最后，我们处理数据——本例中只是用 GZIP 压缩——并把输出发送到 "Processed Data" Output Port。 数据会在这里排队，直到所有数据都准备好离开 Process Group 时才一并释放。

Record Schema

当 Processor 配置为使用 Record Writer 写出列表时，Record 会使用以下 schema（Avro 格式）写出：

{
  "type": "record",
  "name": "lynxusRecord",
  "namespace": "cn.org.acal.lynxus",
  "fields": [
    {
      "name": "filename",
      "type": "string"
    },
    {
      "name": "path",
      "type": "string"
    },
    {
      "name": "directory",
      "type": "boolean"
    },
    {
      "name": "size",
      "type": "long"
    },
    {
      "name": "lastModified",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    },
    {
      "name": "permissions",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "owner",
      "type": [
        "null",
        "string"
      ]
    },
    {
      "name": "group",
      "type": [
        "null",
        "string"
      ]
    }
  ]
}