ListFile

文档中心 › 组件索引 › ListFile

ListFile Processor

类型全限定名:cn.org.acal.lynxus.processors.standard.ListFile

所属 Bundle:cn.org.acal.lynxus:lynxus-standard-nar:2.11.8

描述

从输入目录中检索文件列表。 对于列出的每个文件,创建一个代表该文件的 FlowFile,以便可以与 FetchFile 一起提取该文件。 当“输入目录位置”设置为“远程”时,该处理器设计为仅在集群中的主节点上运行。 如果主节点发生变化,新的主节点将从前一个节点停止的位置继续,而不会复制所有数据。 当“输入目录位置”为“本地”时,“执行”模式可以是任何模式,并且不会发生同步。 与 GetFile 不同,该处理器不会从本地文件系统中删除任何数据。

标签

filegetlistingestsourcefilesystem

属性

下表中,加粗的属性为必填项;其余为可选项。表格同时标注默认值、是否支持表达式语言,以及是否为敏感属性(值会被加密)。

名称默认值可选值描述
Input Directory要从中拉取文件的输入目录
支持表达式语言:是(仅使用环境/系统变量求值)
Listing Strategytimestamps
  • 跟踪时间戳 — 此策略跟踪已列取实体的最新时间戳以确定新/更新的实体。由于它只跟踪少量时间戳,因此可以高效管理列取状态。但是,任何新添加或更新的实体如果时间戳早于跟踪的最新时间戳,则无法被此策略拾取。例如,当带有旧时间戳的文件被复制或移动到目标目录且未更新其最后修改时间戳时,就可能发生这种情况。在列取运行时同时写入多个子目录时也可能遗漏文件。
  • 跟踪实体 — 此策略跟踪最新 '实体跟踪时间窗口' 内所有已列取实体的信息以确定新/更新的实体。此策略可以拾取具有旧时间戳的实体,这些实体可能在 '跟踪时间戳' 策略中被遗漏。即使在列取运行时同时写入多个子目录时也能正常工作。但是需要额外的 DistributedMapCache 控制器服务,并且使用更多的 JVM 堆内存。有关其工作方式的更多详细信息,请参阅 '实体跟踪时间窗口' 属性的描述。
  • 不跟踪 — 此策略在没有任何跟踪的情况下列取实体。每次执行此处理器时,同一实体都会被列取。建议更改默认的运行计划值。与持久化状态相关的任何属性都将被忽略。
指定如何确定新的/更新的实体。 有关详细信息,请参阅每个策略的说明。
Recurse Subdirectoriestrue
  • true
  • false
指示是否列出目录子目录中的文件
Record Writer控制器服务 API:
RecordSetWriterFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8
指定用于创建列表的记录写入器。如果未指定,将为每个列出的实体创建一个 FlowFile。如果指定了记录写入器,所有实体将写入单个 FlowFile,而不是将属性添加到各个 FlowFile。
Input Directory LocationLocal
  • 本地 — 输入目录位于本地磁盘上。状态将本地存储在集群中的每个节点上。
  • 远程 — 输入目录位于远程系统上。状态将在集群中存储,以便可以仅在主节点上执行列取,并且如果主节点更改,另一个节点可以从上一个节点停止的地方继续。
指定输入目录的位置。这用于确定状态应本地存储还是跨集群存储。
File Filter[^\.].*仅选取名称与给定正则表达式匹配的文件
Path Filter当 Recurse Subdirectories 为 true 时,仅扫描路径匹配给定正则表达式的子目录
Include File Attributestrue
  • true
  • false
指定是否应包含文件的最后修改时间和所有者等信息作为 FlowFile 属性。根据所使用的文件系统,收集此信息可能开销很大,因此应禁用。对于远程文件共享尤其如此。
Minimum File Age0 sec提取文件必须达到的最低年龄; 任何早于该时间(根据上次修改日期)的文件都将被忽略
Maximum File Age拉取文件必须达到的最长期限; 任何早于该时间(根据上次修改日期)的文件都将被忽略
Minimum File Size0 B拉取文件必须达到的最小大小
Maximum File Size可以拉取的文件的最大大小
Ignore Hidden Filestrue
  • true
  • false
指示是否应忽略隐藏文件
目标系统时间戳精度auto-detect
  • 自动检测 — 基于候选条目的时间戳自动确定时间单位。请注意,如果没有条目具有精确精度的时间戳,此选项可能会不必要地增加列取时间。例如,即使目标系统支持毫秒,如果所有条目的时间戳都没有毫秒部分,如 '2017-06-16 09:06:34.000',则其精度被确定为 '秒'。
  • 毫秒 — 如果目标系统支持毫秒,此选项可提供从条目可用到被列取的最小延迟。
  • 秒 — 对于不具有毫秒精度但具有秒级精度的目标系统。
  • 分钟 — 对于只支持分钟精度的目标系统。
指定目标系统的时间戳精度。由于此处理器使用实体的时间戳来决定应该列出哪些内容,因此使用正确的时间戳精度至关重要。
Entity Tracking State Cache控制器服务 API:
DistributedMapCacheClient
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8
已列出的实体存储在指定的缓存中, 以便该处理器能够在 Lynxus 重启或主节点变更后恢复列举。 'Tracking Entities' 策略需要跟踪最近 'Tracking Time Window' 内所有已列出实体的信息。 为支持大量实体,该策略使用 DistributedMapCache 而非托管状态。 缓存键格式为 'ListedEntities::{processorId}(::{nodeId})'。 如果按节点跟踪已列出的实体,则会添加可选的 '::{nodeId}' 部分以分别管理状态。 例如,集群范围缓存键 = 'ListedEntities::8dda2321-0164-1000-50fa-3042fe7d6a7b', 单节点缓存键 = 'ListedEntities::8dda2321-0164-1000-50fa-3042fe7d6a7b::lynxus-node3'。 存储的缓存内容是经过 Gzip 压缩的 JSON 字符串。 当目标列举配置发生变化时,缓存键将被删除。 由 'Tracking Entities' 策略使用。
Entity Tracking Time Window3 hours指定该处理器应跟踪已列出实体的时长。 'Tracking Entities' 策略可以选取时间戳位于指定时间窗口内的任何实体。 例如,若设为 '30 minutes',则该处理器运行时,最近 30 分钟内带有时间戳的任何实体都将成为列举目标。 当满足以下任一条件时,已列出的实体被视为 'new/updated' 并发出一个 FlowFile: 1. 不存在于已列出的实体中, 2. 时间戳比缓存的实体更新, 3. 大小与缓存的实体不同。 如果某个缓存实体的时间戳早于指定的时间窗口,该实体将从缓存的已列出实体中移除。 由 'Tracking Entities' 策略使用。
支持表达式语言:是(仅使用环境/系统变量求值)
Entity Tracking Initial Listing Targetall
  • 跟踪时间窗口 — 在首次列取活动中忽略时间戳早于指定 '跟踪时间窗口' 的实体。
  • 所有可用 — 不考虑实体时间戳,在首次列取活动中将列取所有现有实体。
指定应如何处理初始列举。 由 'Tracking Entities' 策略使用。
Entity Tracking Node Identifier${hostname()}配置的值将追加到缓存键中, 以便在跟踪状态作用域为 LOCAL 时, 可以按每个 Lynxus 节点(而非整个集群)跟踪列举状态。 由 'Tracking Entities' 策略使用。
支持表达式语言:是(仅使用环境/系统变量求值)
Track Performancefalse
  • true
  • false
指定处理器是否应跟踪磁盘访问操作的性能。如果为 true,将记录所有磁盘访问,包括正在访问的文件、获取的信息以及所需时间。然后以 DEBUG 级别定期记录这些信息。虽然数据量将被限制,但此选项仍可能消耗大量堆内存(由'最大跟踪文件数'属性控制),但如果性能下降,它对故障排除非常有用。
Maximum Number of Files to Track100000如果 'Track Performance' 属性设置为 'true',此属性指定应保留性能指标的最大文件数。较小的值将减少堆使用,而较大的值可能提供更准确的磁盘访问操作性能洞察
支持表达式语言:是(仅使用环境/系统变量求值)
Max Disk Operation Time10 secs任何单个磁盘操作预期的最大时间。如果任何磁盘操作超过此时间,将为每个超过此时间的操作生成警告公告。
支持表达式语言:是(仅使用环境/系统变量求值)
Max Directory Listing Time3 mins列出任何单个目录预期的最大时间。如果 'Input Directory' 属性指定的目录的列出,或任何子目录的列出(如果 'Recurse' 设置为 true)超过此时间,将为每个超过此时间的目录列出生成警告公告。
支持表达式语言:是(仅使用环境/系统变量求值)

关系 (Relationships)

名称描述自动终止
success所有收到的 FlowFiles 都会成功路由否

写入的 FlowFile 属性

名称描述
filename从文件系统读取的文件名。
path路径设置为文件目录相对于 Input Directory 属性的文件系统上的相对路径。例如,如果 Input Directory 设置为 /tmp,则从 /tmp 拾取的文件的路径属性设置为"/"。如果 Recurse Subdirectories 属性设置为 true 且从 /tmp/abc/1/2/3 拾取文件,则路径属性设置为"abc/1/2/3/"。
absolute.pathabsolute.path 设置为文件目录在文件系统上的绝对路径。例如,如果 Input Directory 属性设置为 /tmp,则从 /tmp 拾取的文件的路径属性设置为"/tmp/"。如果 Recurse Subdirectories 属性设置为 true 且从 /tmp/abc/1/2/3 拾取文件,则路径属性设置为"/tmp/abc/1/2/3/"。
file.owner拥有文件系统中该文件的用户
file.group文件系统中拥有该文件的组
file.size文件系统中文件的字节数
file.permissions文件系统中文件的权限。格式为所有者 3 个字符、组 3 个字符、其他用户 3 个字符。例如 rw-rw-r--
file.lastModifiedTime文件系统中文件的最后修改时间戳,格式为 'yyyy-MM-dd'T'HH:mm:ssZ'
file.lastAccessTime文件系统中文件的最后访问时间戳,格式为 'yyyy-MM-dd'T'HH:mm:ssZ'
file.creationTime文件系统中文件的创建时间戳,格式为 'yyyy-MM-dd'T'HH:mm:ssZ'

状态管理

作用域描述
LOCAL、CLUSTER执行文件列表后,存储最新文件的时间戳。这允许处理器在下次运行时仅列出在此日期之后添加或修改的文件。状态是以本地还是集群范围存储取决于 <Input Directory Location> 属性的值。

受限 (Restricted)

该组件不受限。

输入要求

该组件不允许有传入连接。

另请参阅

详细说明

ListFile

ListFile 会对配置目录中遇到的所有文件执行列表(listing)操作。常见的用例大致可分为两类。

流式用例

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

批处理用例

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

为解决这一问题,ListFile 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。

在这个流程中,我们用 ListFile 对目录做列表操作。该 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。因为我们要用 FetchFile 获取数据,所以需要把 文件的 filename 和 path 作为 FlowFile Attribute。实现方式有几种,最简单的机制是使用 PartitionRecord Processor。该 Processor 配置了一个能够读取 ListFile 所写数据的 Record Reader (本例为 CSV Reader),并配置了两个额外的用户自定义属性:

  • absolute.path: /path
  • filename: /filename

这样,进入 PartitionRecord Processor 的每条记录都会被拆分为一个独立的 FlowFile(因为 "path" 与 "filename" 字段的组合对每条记录都是唯一的),"filename" 和 "path" 记录字段会成为 FlowFile 上的 Attribute(分别使用 "absolute.path" 和 "filename" 作为属性名)。FetchFile 使用默认配置,恰好引用这些 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"
      ]
    }
  ]
}