0624-6.2.0-NiFi处理器介绍与实操

温馨提示：如果使用电脑查看图片不清晰，可以使用手机打开文章单击文中的图片放大查看高清原图。

Fayson的github： https://github.com/fayson/cdhproject

提示：代码块部分可以左右滑动查看噢

1

文档编写目的

Fayson在前面的文章介绍了什么是NiFi，参考《0622-什么是Apache NiFi》。同时对如何在CDH中使用Parcel安装CFM做了介绍，参考《0623-6.2.0-如何在CDH中安装CFM》。本文会首先对NiFi的使用做一下简单的介绍，然后对处理器（Processor）进行详细介绍。

• 测试环境

1.Redhat7.4

2.CM/CDH6.2

3.CFM1.0

4.NiFi1.9

2

NiFi操作简介

1.使用http://<NiFi_HostName>:8080/nifi登录到NiFi画面，可以看到一个用于编排数据流的空白画布。

2.UI有多种工具可用于创建和管理您的第一个数据流：

3.全局菜单包含以下选项：

3

NiFi处理器介绍

3.1

增加一个处理器(Processor)

1.我们现在可以通过在画布中添加Processor来开始创建数据流。 为此，请从屏幕左上角拖动“处理器”图标(

)到画布中间，这时会弹出一个对话框，允许我们选择要添加的处理器：

这里提供了快300种Processor。当开发人员创建Processor时，开发人员会为该处理器分配“tags”，可以认为是处理器的关键字。你可以通过在右上角的“filter”框中输入tag或者处理器的名称来进行过滤。假设我们想把本地磁盘的文件导入NiFi，可以输入关键字“file”，NiFi默认提供了一些处理文件的不同处理器，或者也可以输入“local”来快速缩小列表范围。当你选择了一个处理器后，在对话框底部可以看到处理器的简要说明，告诉你处理器的具体功能。GetFile处理器的描述告诉我们它将数据从本地磁盘拉入NiFi，然后删除本地文件。然后我们可以双击处理器，或者单击选择它，然后点击“Add”按钮，这样处理器就会被添加到画布中。

3.2

配置一个处理器

1.现在我们已经添加了GetFile处理器，可以右键单击处理器并选择Configure来配置它。这时会弹出一个对话框，选择Properties选项卡，会列出许多属性。可用的属性取决于处理器的类型，并且每种类型通常都不同，粗体属性是必需属性。在配置完所有必需属性之前，无法启动处理器。如果我们将目录名（Input Directory）设置为“/data/nifi”,注意这里配置的是绝对路径，这样NiFi就会开始采集该目录的任何数据。我们可以选择为此处理器配置多个不同的属性。如果不确定特定属性的作用，我们可以将鼠标悬停在属性名称旁边的“帮助”（

）图标上，以便阅读该属性的描述。此外，将鼠标悬停在“帮助”图标上时提示将提供该属性的默认值（如果存在）。

3.3

连接处理器

1.每个处理器都有一组定义的“Relationships”，它能够将数据发送到这些关系。处理器完成FlowFile的处理后，会将其传输到其中一个Relationships。这允许用户根据处理的结果配置如何处理FlowFiles。例如，许多处理器定义了两个关系：success和failure。如果处理器能够成功处理数据，则将数据路由到下一个节点，否则如果处理器由于某种原因无法处理数据，则会以完全不通的方式路由到别的地方。或者根据实际情况，也可以将2个relationships都路由到相同的地方。

2.现在我们已经添加并配置了我们的GetFile处理器并应用了配置，我们可以在处理器的左上角看到一个警告图标（

），表示处理器未处于有效状态。

3.将鼠标悬停在此图标上，我们可以看到尚未定义success的relationship。意味着我们没告诉NiFi对于处理器成功处理的数据应该转移到哪里。

4.为了解决这个问题，让我们按照上面的相同步骤添加另一个可以连接GetFile处理器的处理器。 但是，这一次，我们只需记录FlowFile存在的属性。 为此，我们将添加一个LogAttributes处理器。

5.我们现在可以将GetFile处理器的输出发送到LogAttribute处理器。将鼠标悬停在GetFile处理器上，处理器的中间会显示连接图标。

6.我们可以将此图标从GetFile处理器拖到LogAttribute处理器。这时会弹出一个对话框用于选择我们要为此连接包含哪些Relationships。因为GetFile只有一个关系，success，它会自动为我们选择。

7.单击“Settings ”选项卡提供了一些用于配置此连接的行为方式的选项：

8.我们可以给这个Connection取一个名字，否则，连接名称会基于所选的Relationships。我们还可以设置数据的到期时间。 默认情况下，它设置为“0秒”，表示数据永不过期。但是，我们可以更改该值，以便当此Connection中的数据达到特定时间时，它将自动删除（并将创建相应的EXPIRE Provenance事件）。

9.背压阈值允许我们指定队列到达多少时，不再允许源处理器运行。这可以让我们应对一个处理器生产数据的速度比下一个处理器消费数据要快的情况。如果在整个过程中为每个连接配置了背压，则将数据引入系统的处理器最终会因为背压限制会停止引入新数据，以便我们的系统能够恢复。

10.最后，你在右侧还可以看到Prioritizers。这允许我们控制如何排序此队列中的数据。你可以将Prioritizers从 "Available prioritizers" 列表中拖拽到 "Selected prioritizers" 列表中以激活优先级排序器。如果激活了多个Prioritizers，默认会使用排在第一位的Prioritizer，但如果根据Prioritizer判断2个FlowFiles是相等的，则将使用第二个Prioritizer。

11.对于本次的示例，我们只需单击“Add”即可将“Connection”添加到图表中。这是我们会看到Alert图标已经更改为Stopped。

12.但是，LogAttribute处理器现在无效，因为它的success的Relationship尚未连接到任何内容。让我们通过设置LogAttribute处理器将成功的数据路由到 "Auto Terminated”，这样NiFi会当FlowFile处理完成后“drop”掉数据。右键点击LogAttribute处理器，选择“configure”，在“Settings”选项卡的右侧，选中success Relationship复选框表示Auto Terminated掉数据。点击“APPLY”将关闭对话框并显示两个处理器现在都已停止。

3.4

启动和停止处理器

1.此时，我们的画布上有两个处理器，但没有发生任何事情。为了启动处理器，我们可以单独单击每个处理器，然后右键单击并选择“Start”菜单项。

2.或者，我们可以选择第一个处理器，然后在选择其他处理器的同时按住Shift键以选择两者。 然后，我们可以右键单击并选择“Start”菜单项。

3.或者，我们可以选择第一个处理器，然后在选择其他处理器的同时按住Shift键以选择两者。然后从Operate palette中点击“Start”图标。

4.启动后，处理器左上角的图标将从停止的图标更改为正在运行的图标。

5.然后你可以通过Operate palette中的“Stop”图标，或者右键菜单中的“Stop”菜单项来停止处理器。

6.处理器启动后，我们无法再配置它。当我们右键点击处理器，则只能选择查看配置。为了配置处理器，我们必须首先停止处理器并等待可能正在执行的任何任务完成。当前正在执行的任务数显示在处理器的右上角附近，但如果当前没有任务，则不会显示任何内容。

3.5

获得关于更多处理器信息

由于每个处理器都能够暴露多个不同的Properties和Relationships，因此记住每个处理器的所有不同部分的工作可能很困难。要解决此问题，您可以右键单击处理器并选择“Usage”菜单项。这将为您提供处理器的使用信息，例如处理器的描述，可用的不同Relationships，何时使用不同的Relationships，处理器及其文档公开的Properties，以及哪些FlowFile Attributes（如果有的话）预期会传入FlowFiles，或者哪些Attributes（如果有的话）被添加到传出的FlowFiles中。

至此，NiFi处理器介绍完毕。

参考：

https://nifi.apache.org/docs.html

提示：代码块部分可以左右滑动查看噢

为天地立心，为生民立命，为往圣继绝学，为万世开太平。 温馨提示：如果使用电脑查看图片不清晰，可以使用手机打开文章单击文中的图片放大查看高清原图。