处理器类别 | 学习笔记

开发者学堂课程【NiFi 知识精讲与项目实战（第一阶段）：处理器类别】学习笔记，与课程紧密联系，让用户快速学习知识。

课程地址：https://developer.aliyun.com/learning/course/705/detail/12518

处理器类别

内容介绍

一、数据转换

二、路由和调解

三、数据库访问

四、属性提取

五、系统交互

六、数据提取

七、数据出口/发送数据

八、分裂和聚合

九、HTTP

本节课学习 NiFi 所拥有处理器的划分，它是根据其功能进行划分。

一、数据转换

1.CompressContent ：压缩或解压

2.ConvertCharacterSet ：将用于编码内容的字符从一个字符集转换成另一个字符集，比如 JBK 转换成 UTF 8 。

3. EncryptContent ：加密或解密

4.ReplaceText ：使用正则表达式修改文本内容，修改成所需要的数据

5.TransformXml ：应用 XSLT 转换成 XML 内容

6.JoltTransformjSON ：应用 JOLT 规范来转换到 JSON 内容

二、路由和调解

1.ControlRate ：限制流程中数据流经某部分的速率，来保证服务器不会崩溃。

2.DetectDuplicate ：根据一些用户定义的标准去监视发现重复的 FlowFiles 。通常与 HashContent 一起使用，当发现重复的 FlowFiles ，对它进行 Hash 处理。

3.DistributeLoad ：通过只将一部分数据分发给每个用户定义的关系来实现负载平衡或数据抽样，来保证服务的可用和处理的效率。

4.MonitorActivity ：当用户定义的时间段过去而没有任何数据流经此节点时发送通知。（可选）在数据流恢复时发送通知。

5.RouteOnAttribute ：根据 FlowFile 包含的属性路由 FlowFile ，它与 RouteOnContent 相对应。

6.ScanAttribute ：扫描 FlowFile 上用户定义的属性集，检查是否有任何属性与用户定义的字典匹配。

7.RouteOnContent ：根据 FlowFile 的内容是否与用户自定义的正则表达式匹配。如果匹配，则 FlowFile 将路由到已配置的关系。

8.ScanContent ：在流文件的内容中搜素用户定义字典中存在的术语，并根据这些术语的存在或不存在来路由。字典可以由文本条目或二进制条目组成。

9.ValidateXml ：以 XML 模式验证 XML 内容；根据用户定义的 XML  Schema ，判断 FlowFile 的内容是否有效，进而来路由 FlowFile 。

三、数据库访问

1.ConvertJSONToSQL ：将 JSON 文档转换为 SQL  INSERT 或 UPDATE 命令，然后可以将其传递给 PutSQL  Processor ，就可以向数据库中进行数据的插入或更新。

2.ExecuteSQL ：执行用户定义的 SQL  SELECT 命名，结果为 Avro 格式的 FlowFile 。它查询出的数据为 Avro 格式，需要通过其他转换器将 Avro 转换成所需要的格式。

3.PutSQL ：通过执行 FlowFile 内容定义的 SQL  DDM 语句来更新数据库。

4.SelectHiveQL ：它是用来操作 Hive 数据库的，对 Apache  Hive 数据库执行用户定义的 HiveQL  SELECT 命令，结果为 Avro 或 CSV 格式的 FlowFile ，在 NiFi 中 Avro 格式使用较多。

5.PutHiveSQL ：与 PutSQL 有一定区别， PutSQL 一般来说是针对关系型数据库，而 PutHiveSQL 针对 Hive 数据库，通过执行 FlowFile 内容定义的 HiveQL  DDM 语句来更新 Hive 数据库。

四、属性提取

1.EvaluateJsonPath ：用户提供 JSONPath 表达式（类似于 XPath ，用于 XML 解析／提取），然后根据 JSON 内容评估

这些表达式，用结果值替换 FlowFile 内容或将结果值提取到用户自己命名的 Attribute 中，此关键点是使用 JSONPath 表达式。

2.EvaluateXPath ：用户提供 XPath 表达式，然后根据 XML 内容评估这些表达式，用结果值替换 FlowFile 内容或将结果值提取到用户自己命名的 Attribute 中。

3.EvaluateXQuery ：用户提供 XQuery 查询，然后根据 XML 内容评估此查询，用结果值替换 FlowFile 内容或将结果值提取到用户自己命名的 Attribute 中。

4.ExtractText ：用户提供一个或多个正则表达式，然后根据 FlowFile 的文本内容对其进行评估，然后将结果值提取到用户自己命名的 Attribute 中。

5.HashAttribute ：对用户定义的现有属性列表的串联进行 hash 。

6.HashContent ：对 FlowFile 的内容进行 hash ，并将得到的 hash 值添加到 Attribute 中，这里的 Attribute 只有一个 Hash 值。

7.IdentifyMimeType :评估 FlowFile 的内容，以确定 FlowFile 封装的文件类型。此处理器能够检测许多不同的 MIME 类型，例如图像、文字处理器文档、文本和压缩格式，仅举几例，它都可以将其识别。

8.UpdateAttribute :向 FlowFile 添加或更新任意数量的用户定义的属性，这对于添加静态的属性值以及使用表达式语言动态计算出来的属性值非常有用。该处理器还提供”高级用户界面（ Advanced User  Interface ）”，允许用户根据用户提供的规则有条件地去更新属性。

五、系统交互

1.ExecuteProcess ：运行用户自定义的操作系统命令。进程的 StdOut 被重定向，以便 StdOut 的内容输出为 FlowFile 的内容。此处理器是源处理器（不接受数据流输入，没有上游组件，将脚本执行以后打印出的数据发送到 FlowFile 中）﹣其输出预计会生成新的 FlowFile ，并且系统调用不会接收任何输入。它的特点是只能实行没有参数的命令。如果要为进程提供输入，请使用 ExecuteStreamCommand  Processor 。

2.ExecuteStreamCommand ：运行用户定义的操作系统命名。 FlowFile 的内容可选地流式传输到进程的 Stdin 。 StdOut 的内容输出为 FlowFile 的内容。此处理器不能用作源外理器﹣必须传入 FlowFiles 才能执行。不需要通过流文件的数据来获取、拼接。

六、数据提取

1.GetFile ：将文件内容从本地磁盘（或网络连接的磁盘）流式传输到 NiFi ，然后将磁盘中原始数据删除掉。此外理器应将文件从一个位置移动到另一个位置，而不是用于复制数据，主要是剪切。

2.GetFTP ：通过 FTP 将远程文件的内容下载到 NiFi 中，然后删除原始文件。此处理器应将文件从一个位置移动到另一个位置，而不是用于复制数据。

3.GetSFTP ：通过 SFTP 将远程文件的内容下载到 NiFi 中，然后删除原始文件。此处理器应将文件从一个位置移动到另一个位置，而不是用于复制数据。

4.GetJMSQueue ：从 JMS 队列下载消息，并根据 MS 消息的内容创建 FlowFile 。可选地， JMS 属性也可以作为属性复制。

5.GetJMSTopic ：从 JMS 主题下载消息，并根据 JMS 消息的内容创建 FlowFile 。可选地， JMS 属性也可以作为属性复制。此处理器支持持久订阅和非持久订阅。

6.GetHTTP ：将基于 HTTP 或 HTTPS 的远程 URL 的请求内容下载到 NiFi 中。处理器将记住 ETag 和 Last - Modified  Date ，以确保不会持续摄取数据。

7.ListenHTTP ：主要进行监听，启动 HTTP （或 HTTPS ）服务器并侦听传入连接。对于任何传入的 POST 请求，请求的内容将作为 FlowFile 写出，并返回200响应。

8.ListenUDP ：侦听传入的 UDP 数据包，并为每个数据包或毎个数据包创建一个 FlowFile （取决于配置），并将 FlowFile 发送到＂ success ＂。

9.GetHDFS ：监视 HDFS 中用户指定的目录。每当新文件进入 HDFS 时，它将被复制到 NiFi 并从 HDFS 中删除。它与 GetFile 和 GetFTP 类似，也是执行剪切状态。此处理器应将文件从一个位置移动到另一个位置，而不是用于复制数据。如果在集群中运行，此处理器需仅在主节点上运行。要从 HDFS 复制数据并使其保持原状，或者从群集中的多个节点流式传输数据，请参阅 ListHDFS 处理器。

10.ListHDFS / FetchHDFS : 两者配合使用，ListHDFS 监视 HDFS 中用户指定的目录，并发出一个 FlowFile ，其中包含它遇到的每个文件的文件名。然后，它通过分布式缓存在整个 NiFi 集群中保持此状态。然后可以在集群中，将其发送到 FetchHDFS 处理器，后者获取这些文件的实际内容并发出包含从 HDFS 获取的内容的 FlowFiles 。 ListHDFS 读取文件中所有的文件名， FetchHDFS 将文件中所有文件名的内容读取出来，这样的配合就可以将某一个目录下所有的文件名以及文件的内容获取到。

11.GetKafka ：从 Apache  Kafka 获取消息，特別是0.8. x 版本。消息可以作为每个消息的 FlowFile 发出，也可以使用用户指定的分隔符一起进行批外理。

12.GetMongo ：对 MongoDB 执行用户指定的查询，并将内容写入新的 FlowFile 。

七、数据出口/发送数据

1.PutEmail ：向配置的收件人发送电子邮件。 FlowFile 的内容可选择作为附件发送。

2.PutFile ：将 FlowFile 的内容写入本地（或网络连接）文件系统上的目录。

3.PutFTP ：将 FlowFile 的内容复制到远程 FTP 服务器。

4.PutSFTP ：将 FlowFile 的内容复制到远程 SFTP 服务器。

5.PutJMS ：将 FlowFile 的内容作为 JMS 消息发送到 MS 代理，可选择将 Attributes 添加 JMS 属性。

6.PutSQL ：将 FlowFile 的内容作为 SQL  DDL 语句( INSERT ，UPDATE 或 DELETE ）执行。 FlowFile 的内容必须是具有 SQL 语句。属性可以用作参数， FlowFile 的内容可以是参数化的 SQL 语句，以避免 SQL 注入攻击。

7.PutKafka ：将 FlowFile 的内容作为消息发送到 Apache  Kafka ，特别是0.8. x 版本。 FlowFile 可以作为单个消息或分隔符发送，例如可以指定换行符，以便为单个 FlowFile 发送许多消息。

8.PutMongo ：将 FlowFile 的内容作为 INSERT 或 UPDATE 发送到 Mongo 。

八、分裂和聚合

1.SplitText : SplitTex 接收单个 FlowFile ，其内容为文本，并根据配置的行数将其拆分为1个或多个 FlowFiles 。例如，可以将处理器配置为将 FlowFile 拆分为多个 FlowFile ，每个 FlowFile 只有一行，是根据换行符进行拆分的，或者是 Tab 符号。

2.SplitJson ：允许用户将包含数组或许多子对象的 JSON 对象拆分为每个 JSON 元素的 FlowFile 。

3.SplitXml ：允许用户将 XML 消息拆分为多个 FlowFiles ，每个 FlowFiles 包含原始段。这通常在多个 XML 元素与＂ wrapper "元素连接在一起使用。然后，此处理器允许将这些元素拆分为单独的 XML 元素。

4.UnpackContent ：解压缩不同类型的存档格式，例如 ZIP 和 TAR 。然后，归档中的每个文件都作为单个 FlowFile 传输。

5.SegmentContent ：根据某些已配置的数据大小将 FlowFile 划分为可能的许多较小的 FlowFile 。不对任何类型的分界符执行拆分，而是仅基于字节偏移执行拆分。这是在传输 FlowFiles 之前使用的，以便通过并行发送许多不同的部分来提供更低的延迟。而另一方面， MergeContent 处理器可以使用碎片整理模式重新组装这些

FlowFiles 。

6.MergeContent ：此处理器负责将许多被拆分后的 FlowFiles 合并到一个 FlowFile 中。可以通过将其内容与可选的页眉，页脚和分界符连接在一起，或者通过指定存档格式（如 ZIP 或 TAR ）来合并 FlowFiles 。 FlowFiles 可以根据公共属性进行分箱（ binned ），或者如果这些流是被其他组件拆分的，则可以进行“碎片整理 ( defragmented )” 。根据元素的数量或 FlowFiles 内容的总大小（每个 bin 的最小和最大大小是用户指定的）并且还可以配置可选的 Timeout 属性，即 FlowFiles 等待其 bin 变为配置的上限值最大时间。

7.SplitContent ：将单个 FlowFile 拆分为可能的许多 FlowFile 。类似于 SegmentContent ，但易，对于 SplitContent ，不会在任意字节边界上执行拆分，而是指定要拆分内容的字节序列。

九、HTTP

1.GetHTTP ：将基于 HTTP 或 HTTPS 的远程 URL 的内容下载到 NiFi 中。处理器将记住 ETag 和 Last - Modified Date ,以确保不会持续摄取数据。

2.ListenHTTP ：启动全新的 HTTP （或 HTTPS ）服务器并侦听传入连接。对于任何传入的 POST 请求，请求的内容将作为 FlowFile 写出，并返回200响应。

3.InvokeHTTP ：执行用户配置的 HTTP 请求。此处理器比 GetHTTP 和 PostHTTP 更通用，但需要更多配置。此处理器不能用作源处理器，并且需要具有传入的 FlowFiles 才能被触发以执行其任务。

4.PostHTTP ：执行 HTTP  POST 请求，将 FlowFile 的内容作为消息正文发送。这通常与 ListenHTTP 结合使用，以便在无法使用 s2s 的情况下在两个不同的 NiFi 实例之间传输数据（例如，节点无法直接访问并且能够通过 HTTP 进行通信时代理）。注意：除了现有的 RAW 套接字传输之外， HTTP 还可用作s2s传输协议。它还支持 HTTP 代理。建议使用 HTTP  s2s ，因为它更具可扩展性，并且可以使用具有更好用户身份验证和授权的输入／输出端口的方式来提供双向数据传输。

5.HandleHttpRequest / HandleHttpResponse : HandleHttpRequest  Processor 是一个源处理器，与 ListenHTTP 类似，启动嵌入式 HTTP ( S ）服务器。但是它不会向客户端发送响应（比如200响应），它需要等整个流程结束后进行返回。相反，流文件是以 HTTP 请求的主体作为其内容发送的，其他处理器会对接收到的参数进行处理转换计算，所有典型 servlet 参数、头文件等的属性作为属性。然后， HandleHttpResponse 能够在 FlowFile 完成外理后将响应发送回客户端，这些外理器总是彼此结合使用形成一个全新的 ListenHTTP，并允许用户在 NiFi 中可视化地创建 Web 服务。这对于将前端添加到非基于 Web 的协议或围绕已经由 NiFi 执行的某些功能（例如数据格式转换）添加简单的 Web 服务特别有用。