为了能够创建一个高效的NiFi数据数据流程，我们需要了解有哪些Processor类型可以使用。每个新的NiFi版本Processor数量都会增加，当前NiFi版本为1.12.1,内置了288个类型的Proccessor，这些Processor提供从多个不同系统接收数据、路由、转换、处理、拆分和聚合数据以及将数据分发到多个系统的功能。

按照功能分类：

image.png

. procesor 分类说明

1.1 数据摄取

• GetFile: 此“Processor”监控本地磁盘或者网络连接磁盘中的文件文件夹，读取文件内容封装为一个FlowFile，文件的属性将会转化为FlowFile的属性。默认情况下源文件会被删除（也可选择保留）。会忽略没有读取权限的文件。
• GetFTP: 下载FtpServer上的文件，读取文件内容创建FlowFile。读取完成后原始文件会被删除。此processor的目的是移动文件，而不是复制文件。
• GetSFTP: 下载SFtpServer上的文件，读取文件内容创建FlowFile。读取完成后原始文件会被删除。此processor的目的是移动文件，而不是复制文件。
• GetJMSQueue: 由ActiveMQ JMS消息队列读取一条消息，并更具消息内容创建FlowFile。可以选择是否将JMS属性复制为FlowFile的属性。
• GetJMSTopic: 由ActiveMQ JMS消息队列的topic读取一条消息，并更具消息内容创建FlowFile。可以选择是否将JMS属性复制为FlowFile的属性。支持持久订阅和非持久订阅。
• GetHTTP: 又一个http或者https地址下载内容并创建FlowFile加载到NiFi。这个Processor会记录ETag和Last-Modified Date，以避免重复的获取数据。
• ListenHTTP: 启动一个http server，监听连接。任何Post Request的content都会被读取转化为FlowFile，会返回一个200响应给请求端。
• ListenUDP: 监听传入的UDP数据包，并为每个数据包或每一组数据包创建一个FlowFile（取决于配置），并将FlowFile发送到“success”关系。
• GetHDFS: 监视HDFS中用户指定的目录。 每当有新文件进入HDFS时，就会将其复制到NiFi中并从HDFS中删除。 GetHDFS的设计目标是移动文件，而不是复制文件。 如果该GetHDFS在群集中运行，为了从HDFS复制数据并使其保持完整，只能允许在主节点上运行。如果需要从群集中的多个节点流转输数据，则要试用ListHDFS处理器。
• ListHDFS / FetchHDFS: ListHDFS监视HDFS中用户指定的目录，并发出一个FlowFile，其中包含它遇到的每个文件的文件名。然后，它通过分布式缓存在整个NiFi集群中保持这种状态。然后，这些FlowFile可以在集群中分发到FetchHDFS处理器，后者负责获取这些文件的实际内容，并发送包含从HDFS获取的内容的FlowFile。
• FetchS3Object: 从AWS简单存储服务（S3）获取对象的内容。发出的DataFlow包含从S3接收的内容。
• GetKafka: 由kafka(0.8.0版本)取回数据，将每个消息封装为FlowFile也可以将多条封装为一个FlowFlie(需要自定义分隔符)。其他kafka版本有对应的ConsumeKafkaRecord和ConsumeKafka Processor。
• GetMongo: 执行一个用户定义的查询语句，将返回结果写入新的FlowFile中。
• GetTwitter: 允许用户注册一个过滤器来监听Twitter“garden hose”或“Enterprise endpoint”，为接收到的每个tweet创建一个FlowFile。

1.2. 数据转化

• CompressContent: 压缩、解压FlowFile的内容（Content）。
• ConvertCharacterSet: 将内容从一个字符集编码到另一个字符集的字符集
• EncryptContent: 加密、揭秘FlowFile的内容（Content）。
• ReplaceText: 使用正则表达式修改文本内容
• TransformXml: 使用XSLT转化XML内容
• JoltTransformJSON: 使用 JOLT转化JSON（内容）

1.3. 路由与中介

• ControlRate: 控制数据传输到后续Processor的速率
• DetectDuplicate: 根据一些用户定义的条件，监视重复的FlowFiles。 常与HashContent结合使用。
• DistributeLoad: 可以定义1-n个输出关系，每个关系将输出n分之一的数据，可以用来负载均衡或者采样数据。
• MonitorActivity: 当用户定义的时间段过去而没有任何数据通过流中的特定点时，发送通知。 （可选）在数据流恢复时发送通知。
• RouteOnAttribute: 根据FlowFile包含的属性路由数据。
• ScanAttribute: 扫描FlowFile上用户定义的属性集，检查是否有属性与用户定义的词典相匹配，来决定路由。
• RouteOnContent: 搜索FlowFile的内容以查看其是否与任何用户定义的正则表达式匹配。 根据是否匹配来决定FlowFile流向。
• ScanContent: 在FlowFile的内容中搜索用户定义的词典中存在条目，并根据这些条目的存在或不存在进行路由。 该词典可以包含文本条目或二进制条目。
• ValidateXml: 使用用户定义的XMLschema验证FlowFile的内容是否有效来路由FlowFile。

1.4. 数据库访问

• ConvertJSONToSQL: 将json转化为insert或者update sql，发送给"PutSQL Processor"
• ExecuteSQL:执行一个用户定义的sql，将返回结果转化为Avro格式并做为一个FlowFile的content。
• PutSQL: 通过执行由FlowFile的内容定义的sqlddm语句来更新数据库
• SelectHiveQL: 对Apache配置Hive数据库执行用户定义的HiveQL SELECT命令，将结果写入Avro或CSV格式的FlowFile
• PutHiveQL: 通过执行由FlowFile内容定义的HiveQL DDM语句来更新配置的Hive数据库
• PutInfluxDB:将FlowFile内容中的'line protocol'数据插入到influxdb中。

1.5. 属性提取

• EvaluateJsonPath: 用户提供JSONPath表达式（类似于XPath，用于XML解析/提取），然后根据表达式内容对这些json进行计算提取，以替换FlowFile内容或将值提取到用户命名的Attribute中。
• EvaluateXPath: 用户提供XPath表达式，然后根据XML内容对这些表达式进行计算提取，以替换FlowFile内容或将值提取到用户命名的Attribute中。
• EvaluateXQuery: 用户提供一个XQuery查询，然后根据XML内容对该查询进行计算提取，以替换FlowFile内容或将该值提取到用户命名的Attribute中。
• ExtractText: 使用一个或多个正则表达是有FlowFile内容中提取数据，以替换FlowFile内容或将该值提取到用户命名的Attribute中。
• HashAttribute: 针对用户定义的一个属性列表串联执行哈希函数。
• HashContent: 针对FlowFile的内容执行哈希函数，并将哈希值添加为属性。
• IdentifyMimeType: Evaluates the content of a FlowFile in order to determine what type of file the FlowFile encapsulates. This Processor is capable of detecting many different MIME Types, such as images, word processor documents, text, and compression formats just to name a few.
• UpdateAttribute: 向FlowFile添加或更新任意数量的用户定义属性。 这对于添加静态配置的值以及使用表达式语言动态派生属性值很有用。 该处理器还提供了一个“高级用户界面”，允许用户根据用户提供的规则有条件地更新属性。

1.6. 系统交互

• ExecuteProcess: 执行一个系统命令，将命令的标准输出捕获做为新建FlowFile的Content。命令接收任何输出。 例如可以使用“tail”命令补货文件内容。
• ExecuteStreamCommand: 执行一个系统命令，将输入的FlowFile内容做为命令的输入，将命令的输出做为输出FlowFile的内容。

1.7.数据出口/数据发送

• PutEmail: 向配置的邮件地址发送电子邮件。 FlowFile的内容可以选择作为附件发送。
• PutFile: 将FlowFile的内容写入本地（或网络连接）文件系统上的目录中。
• PutFTP: 将FlowFile的内容复制到远程FTP服务器。
• PutSFTP: 将FlowFile的内容复制到远程SFTP服务器。
• PutJMS: 将FlowFile的内容作为JMS消息发送到JMS代理，可以选择基于属性添加JMS属性。
• PutSQL: 以SQL DDL语句（INSERT，UPDATE或DELETE）的形式执行FlowFile的内容。 FlowFile的内容必须是有效的SQL语句。可以将属性用作参数，以便可以将FlowFile的内容参数化为SQL语句，以避免SQL注入攻击。
• PutKafka: 将FlowFile的内容作为消息发送到Apache Kafka，特别是针对0.8.x版本。 FlowFile可以作为单个消息发送，也可以使用分隔符（例如可以指定换行符）发送，以便为单个FlowFile发送许多消息。
• PutMongo: 将FlowFile的内容作为INSERT或UPDATE发送到Mongo。

1.8. 拆分与合并

• SplitText: SplitText接收单个FlowFile，其内容为文本，然后根据配置的行数将其拆分为1个或多个FlowFile。例如，处理器可以配置为将FlowFile拆分为多个FlowFile，每个文件只有1行。
• SplitJson: 允许用户根据JSON元素将由数组或许多子对象组成的JSON对象拆分为FlowFile。
• SplitXml: 允许用户将XML消息拆分为许多FlowFile，每个文件都包含原始文件的一部分。通常在将几个XML元素与“包装”元素结合在一起时使用。然后，此处理器允许将那些元素拆分为单独的XML元素。
• UnpackContent: 解压缩不同类型的存档格式，例如ZIP和TAR。然后，存档中的每个文件都将作为单个FlowFile进行传输。
• MergeContent: 此处理器负责将多个FlowFile合并为一个FlowFile。可以通过将FlowFiles的内容与可选的页眉，页脚和分界符连接在一起，或通过指定存档格式（例如ZIP或TAR）来合并FlowFiles。 FlowFiles可以基于公共属性进行装箱，如果通过其他拆分过程将它们分开，则可以对其进行“碎片整理”。每个bin的最小和最大大小由用户指定，具体取决于元素的数量或FlowFiles内容的总大小，还可以分配一个可选的Timeout，以便FlowFiles仅在特定时间内等待其bin变满多少时间。
• SegmentContent: 根据一些已配置的数据大小，将FlowFile分割为可能较小的FlowFile。不针对任何类型的分界符执行拆分，仅基于字节偏移量执行。在传输FlowFiles之前使用它，以便通过并行发送许多不同的片段来提供较低的延迟。在另一方面，这些FlowFiles然后可以由MergeContent处理器使用碎片整理模式重新组装。
• SplitContent: 将单个FlowFile拆分为可能的多个FlowFile，类似于SegmentContent。但是，使用SplitContent时，不对任意字节边界执行拆分，而是指定了要在其上拆分内容的字节序列。

1.9. http

• GetHTTP: 将基于远程HTTP或HTTPS的URL的内容下载到NiFi中。处理器将记住ETag和上次修改日期，以确保不会重复摄取数据。
• ListenHTTP: 启动HTTP（或HTTPS）服务器并监听传入的连接。对于任何传入的POST请求，将请求的内容将作为FlowFile内容，并返回200响应。
• InvokeHTTP: 执行由用户配置的HTTP请求。该处理器比GetHTTP和PostHTTP具有更多的用途，但是需要更多的配置。该处理器不能用作“源处理器”，并且必须具有传入的FlowFiles才能被触发执行其任务。
• PostHTTP: 执行HTTP POST请求，将FlowFile的内容作为Post的内容。在无法使用s2s的情况下，通常将它与ListenHTTP结合使用，以便在两个不同的NiFi实例之间传输数据。 注意：HTTP可以作s2s传输使用协议以及现有的RAW套接字传输。它还支持HTTP代理。建议使用HTTP Site-to-Site，因为它具有更高的可扩展性，并且可以使用输入/输出端口提供双向数据传输，并具有更好的用户身份验证和授权。
• HandleHttpRequest / HandleHttpResponse: HandleHttpRequest处理器是一个源处理器，它启动类似于ListenHTTP的嵌入式HTTP（S）服务器。但是，它不会向客户端发送响应。相反，将以HTTP请求的主体作为其内容和属性（所有典型Servlet参数，标头等）作为属性来发送FlowFile。然后，在FlowFile处理完毕后，HandleHttpResponse便能够将响应发送回客户端。始终希望这些处理器可以相互结合使用，并允许用户在NiFi中直观地创建Web服务。这对于将前端添加到非基于Web的协议或围绕NiFi已执行的某些功能（例如数据格式转换）添加简单的Web服务特别有用

1.10. amazon web services

• FetchS3Object: 获取存储在Amazon Simple Storage Service（S3）中的对象的内容。 然后将其写入FlowFile的内容。
• PutS3Object: 使用配置的凭证，密钥和存储桶名称将FlowFile的内容写入Amazon S3对象。
• PutSNS: 将FlowFile的内容作为通知发送到Amazon Simple Notification Service（SNS）。
• GetSQS: 从Amazon Simple Queuing Service（SQS）中提取一条消息，并将消息的内容写入FlowFile的内容。
• PutSQS: 将FlowFile的内容作为消息发送到Amazon Simple Queuing Service（SQS）
• DeleteSQS:从Amazon Simple Queuing Service（SQS）中删除一条消息。 可以将其与GetSQS结合使用，以便从SQS接收消息，对其进行一些处理，然后仅在对象成功完成处理后才从队列中删除该对象。

2.操作属性

每个FlowFile创建时都有几个属性，并且属性将在FlowFile的生命周期内发生变化。 属性这个概念提供了三个主要好处：

• 首先，它允许用户在数据流程中制定路由决策，根据属性决定数据流经不同的流程路径。

• 其次，保存了数据元数据，很多Processor需要这些元数据做为参数。以使处理器的配置取决于数据本身。例如，PutFile Processor能够使用属性来知道每个FlowFile的存储位置，而每个FlowFile的目录和文件名属性可能不同。

• 最后，属性为数据提供了非常有价值的上下文。属性可以做为搜索数据的条件，这允许用户搜索与特定条件匹配的出处数据，并且还允许用户在检查出处事件的详细信息时查看此上下文。通过这样做，用户只需浏览一下随内容一起携带的上下文，就能够理解为什么以一种或另一种方式处理数据。

2.1 一般属性(系统属性)

所有FlowFile都会有的属性一个最小的属性集合，一般由NiFi进行维护 :

• filename: 文件名，可以用来把数据保存到本地磁盘或者远程文件系统(nas,hdfs等)。
• path: 路径，存储数据的本地或者远程文件夹
• uuid: 每个FlowFile都有的唯一值，用于在nifi中唯一标识一个FlowFile.
• entryDate: 数据进入NiFi的时间. 是长整数，表示"1970.1.1 00:00:00" 到当前时间点的毫秒数.
• lineageStartDate: 一个FlowFile被复制、合并、拆分都会创建一个或者多个子FlowFile。这些子FlowFile再经过转换就构成了一个血缘链。A此值表示最早的祖先进入系统的日期和时间。 考虑另一种意义，该属性表示FlowFile通过系统的延迟。 该值是一个数字，表示自1970年1月1日午夜（UTC）以来的毫秒数。
• fileSize: 内容的字节数，即文件的大小。

这些属性都是系统生成的不能修改。

2.2 提取属性

NiFi提供了几种开箱即用的Processor，用于从FlowFiles中提取属性。同样，这也是构建自定义Processor的常见的例子。 许多Processor被编写来处理特定的数据格式，并从FlowFile的内容中提取相关信息，创建属性来保存该信息，以便随后就如何路由或处理数据做出决策。

2.3 添加用户自定义属性

除了使Processor能够从FlowFile内容中提取特定信息到属性外，用户还需要在数据流程中的特定位置向每个FlowFile添加用户定义属性。“UpdateAttribute Processor”专为此目的而设计。通过单击“属性”选项卡右上角的“ +”按钮，用户可以在“配置”对话框中向Processor添加新属性。然后提示用户输入属性的名称，然后输入值。对于此UpdateAttribute Processor处理的每个FlowFile，将为会把用户定义的属性添加到FlowFile属性中。属性的名称将与添加的属性的名称相同。属性的值将与属性的值相同，属性的值也可以包含表达语言。这允许基于其他属性修改或添加属性。例如，如果要在处理文件的主机名和日期前加上文件名，则可以通过添加名称为“ filename”和值为“{now():format('yyyy-dd-MM')}-${filename}”的属性来实现，这需要用户熟悉NiFi表达式语言。 除了固定添加一组定义的属性外，UpdateAttribute Processor还具有一个高级UI，该UI允许用户配置一组规则，在这些规则上应添加属性。要使用此功能，请在“配置”对话框的“属性”标签中，点击对话框底部的“高级”按钮。这是提供专门针对此processor定制的UI，其他Processor没有此功能。在此UI内，用户能够配置规则引擎，从本质上讲，指定必须匹配的规则才能将已配置的属性添加到FlowFile。

• UpdateAttribute 属性配置窗口

  
  image.png

• 自定义属性,输入属性名称

  
  image.png

• 自定义属性，配置规则界面

  
  image.png

2.4 属性路由

NiFi最强大的功能之一是能够基于FlowFile的属性来路由它们。执行此操作的主要方法是RouteOnAttribute Processor。该Processor与UpdateAttribute一样，是通过添加用户定义的属性来配置的。通过单击“Processor”的“配置”对话框中“属性”选项卡右上角的“ +”按钮，可以添加任意数量的属性。 每个FlowFile的属性都将与配置的属性进行比较，以确定FlowFile是否满足指定的条件。每个属性的值都应为表达式语言表达式，并返回布尔值。 在计算了针对FlowFile的属性提供的表达语言表达式后，Processor将根据所选的路由策略确定如何路由FlowFile。最常见的策略是“路由到属性名称”策略。选择此策略后，Processor将为每个配置的属性创建一个关系。如果FlowFile的属性满足给定的表达式，则FlowFile的副本将被路由到相应的Relationship。例如，如果我们有一个名称为“ begins-with-r”且值为“ $ {filename：startsWith（'r'）}”的新属性，则任何文件名以字母“ r”开头的FlowFile都将是路由到该关系。所有其他FlowFiles将被路由到“ unmatched”。

2.5 表达式语言 / Using Attributes in Property Values

从FlowFiles的内容中提取属性并添加用户定义的属性时，除非我们有某种使用它们的机制，否则它们对我们作为操作没有多大帮助。 NiFi表达式语言使我们能够在配置流程时访问和操作FlowFile属性值。并非所有处理器属性都允许使用表达式语言，但许多属性都可以使用。为了确定属性是否支持表达式语言，用户可以将鼠标悬停在“帮助”图标（

image.png
）在“Processor configure”对话框的“Propreties”选项卡中。这将提供一个工具提示，显示该属性的描述，默认值（如果有）以及该属性是否支持表达式语言。
image.png
对于支持表达式语言的属性，可以通过在开始的" {uuid}”。如果属性名称以字母以外的其他字符开头，或者包含数字，字母，句点（。）或下划线（_）以外的其他字符，则需要用引号引起来。例如，“ {'My Attribute Name'}”将指向属性“My Attribute Name”。
除了引用属性值外，我们还可以对这些属性执行许多功能和比较。例如，如果我们要检查filename属性是否包含字母'r'而不注意大小写（大写或小写），则可以使用表达式${filename:toLower():contains('r')}。注意这里的功能用冒号分隔。我们可以将任意数量的函数链接在一起以构建更复杂的表达式。同样重要的是，即使我们正在调用filename：toLower（），这也不会改变filename属性的值，而只是为我们提供了一个新的值。
我们还可以将一个表达式嵌入另一个表达式中。例如，如果我们想将“ attr1”属性的值与“ attr2”属性的值进行比较，则可以使用以下表达式来做到这一点：$ {attr1：equals（$ {attr2}）}。
表达式语言包含许多功能，可用于执行路由和处理属性。函数（Functions）用于解析和处理字符串，比较字符串和数字值，处理和替换值以及比较值。 此外，表达式语言文档内置于应用程序中，因此用户可以在键入时轻松查看可用的功能并查看其文档。设置支持表达式语言的属性的值时，如果光标位于表达式语言的开始和结束标记之内，则按关键字上的Ctrl +空格键将弹出所有可用功能，并且会自动完整的功能。单击或使用键盘导航到弹出窗口中列出的功能之一，将显示一个工具提示，其中解释了该功能的作用，期望的参数以及该函数的返回类型。

2.6 表达式语言中的自定义属性

除了使用FlowFile属性外，还可以定义用于表达语言的自定义属性。 自定义属性使在处理和配置数据流时具有更多的灵活性。 例如，可以引用连接，服务器和服务属性的自定义属性。 创建自定义属性后，可以在“ nifi.properties”文件中的“ nifi.variable.registry.properties”字段中标识它们的位置。 更新“ nifi.properties”文件并重新启动NiFi之后，便可以根据需要使用自定义属性。