本开发人员指南旨在为读者提供叻解Apache NiFi扩展如何开发所需的信息并帮助解释开发组件背后的思维过程。它提供了用于开发扩展的API的介绍和说明但是，它没有详细介绍API中嘚每个方法因为本指南旨在补充API的JavaDoc而不是替换它们。本指南还假定读者熟悉Java 7和Apache Maven

本指南由开发人员为开发人员编写。在阅读本指南之前您需要对NiFi和数据流概念有基本的了解。如果没有请参阅 和以熟悉NiFi的概念。

NiFi提供了几个扩展点使开发人员能够为应用程序添加功能以滿足他们的需求。以下列表提供了最常见扩展点的高级描述：

• • 处理器接口是NiFi公开对其属性及其内容的访问的机制 。处理器是用于构成NiFi数據流的基本构建块此接口用于完成以下所有任务：

• • ReportingTask接口是NiFi公开的一种机制，允许将指标监控信息和内部NiFi状态发布到外部端点，例如日誌文件电子邮件和远程Web服务。

• • ControllerService在单个JVM中跨处理器其他ControllerServices和ReportingTasks提供共享状态和功能。示例用例可以包括将非常大的数据集加载到存储器中通过在ControllerService中执行此工作，数据可以加载一次并通过此服务公开给所有处理器，而不是要求许多不同的处理器自己加载数据集

• • FlowFilePrioritizer接口提供了┅种机制，通过该机制可以对队列中的进行优先级排序或排序以便可以按照对特定用例最有效的顺序处理FlowFiles。

• • AuthorityProvide负责确定应授予给定用户的特权和角色（如果有）

处理器是NiFi中使用最广泛的组件。处理器是唯一可以访问以创建删除，修改或检查FlowFiles（数据和属性）的组件

使用Java嘚ServiceLoader机制加载和实例化所有处理器。这意味着所有处理器必须遵守以下规则：

• 处理器必须具有默认构造函数

• 处理器在getRelationships调用其方法时返回用戶指定的所有关系，并且还将返回unmatched关系因为此处理器必须读取和写入内容存储库（这可能相对昂贵），如果预期此处理器将用于非常高嘚数据量则添加允许用户指定是否有用的属性可能是有利的。他们是否关心与任何匹配标准不匹配的数据

  via mit可能不会立即生效。相反這些提交可以一起批处理以提供更高的吞吐量。但是如果处理器在任何时候回滚ProcessSession，则自上次调用以来的所有更改commit都将被丢弃所有“批處理”提交都将生效。这些“批量”提交不会回滚

  在设计处理器时，需要考虑一些重要的设计“开发人员指南”的这一部分将开发人員在创建处理器时应考虑的一些想法放在首位。

  开发处理器（或任何其他组件）时要记住的最重要的概念之一是您正在创建的用户体验偅要的是要记住，作为这样一个组件的开发者您可能对其他人没有的上下文有重要的了解。应始终提供文档以便不熟悉该过程的人员能够轻松使用它。

  在考虑用户体验时同样重要的是要注意一致性非常重要。最好坚持使用标准这适用于处理器名称，属性名称和值關系名称以及用户将体验的任何其他方面。

  简单至关重要！避免添加您不希望用户理解或更改的属性作为开发人员，我们被告知硬编码徝很糟糕但这有时会导致开发人员暴露属性，当要求澄清时告诉用户只保留默认值。这导致混乱和复杂性

  为了制作单个，有凝聚力嘚单元开发人员有时会试图将多个功能组合到一个处理器中。对于处理器期望输入数据采用格式X以便处理器可以将数据转换为格式Y并将噺格式化的数据发送到某些外部服务的情况情况确实如此。

  采用这种方法格式化特定端点的数据然后将数据发送到同一处理器内的该端点有几个缺点：

  • 处理器变得非常复杂，因为它必须执行数据转换任务以及将数据发送到远程服务的任务

  • 如果处理器无法与远程服务通信，它将把数据路由到failure关系在这种情况下，处理器将负责再次执行数据转换如果它再次失败，则翻译再次完成

  • 如果我们有五个不同嘚处理器在传输数据之前将传入数据转换为这种新格式，那么我们就会有大量重复的代码例如，如果架构发生更改则必须更新许多处悝器。

  • 当处理器完成发送到远程服务时将丢弃此中间数据??。中间数据格式可能对其他处理器有用

  为了避免这些问题，并使处理器哽具可重用性处理器应始终坚持“做一件事，做得好”的原则这样的处理器应该分成两个独立的处理器：一个用于将数据从格式X转换為格式Y，另一个处理器用于将数据发送到远程资源

  为了向用户提供一致的外观和感觉，建议处理器遵守标准命名约定以下是使用的标准约定列表：

  • 从远程系统提取数据的处理器名为Get <Service>或Get <Protocol>，具体取决于它们是否通过已知协议（例如GetHTTP或GetFTP）从任意源轮询数据或者它们是否从已知服务提取数据（比如GetKafka）

  • 关系名称是低级的，并使用空格来描述单词

  • 属性名称大写重要单词，就像书名一样

  在创建处理器时，开发人員能够向框架提供有关如何最有效地利用处理器的提示这是通过将注释应用于Processor的类来完成的。可以应用于处理器的注释存在于三个子包Φmit则无法保证数据已安全地存储在NiFi的Content，FlowFile和Provenance Repositories中因此，不适合那些从外部源接收数据提交会话，然后删除远程数据或使用远程资源确认倳务的处理器

• TriggerSerially：当存在此注释时，框架将不允许用户一次调度多个并发线程来执行该onTrigger方法相反，线程数（“并发任务”）将始终设置為1这并没有，但是意味着该处理器不必是线程安全的，因为这是执行的线程onTrigger调用之间可能发生改变

• NiFi在集群时为处理器提供两种执行模式：“主节点”和“所有节点”。虽然在所有节点中运行可提供更好的并行性但已知某些处理器在多个节点中运行时会导致意外行为。例如某些处理器列出或读取远程文件系统中的文件。如果计划在“所有节点”上运行此类处理器则会导致不必要的重复甚至错误。此类处理器应使用此注释应用此批注将限制处理器仅在“主节点”上运行。

• TriggerWhenAnyDestinationAvailable：默认情况下如果任何出站队列已满，NiFi将不会安排处理器運行这允许背压一直应用于处理器链。但是即使其中一个出站队列已满，某些处理器也可能需要运行此注释表示如果任何关系“可鼡”，则应运行处理器如果没有使用该关系的连接已满，则称关系“可用”例如，DistributeLoad Processor使用此批注如果使用“循环”调度策略，则如果任何出站队列已满则处理器将不会运行。但是如果使用“下一个可用”调度策略，

• TriggerWhenEmpty：默认行为是仅在其输入队列至少有一个FlowFile或者处理器没有输入队列（这是典型的“源”处理器）时触发处理器才能运行应用此批注将导致框架忽略输入队列的大小并触发处理器，无论输叺队列中是否有任何数据例如，如果需要触发处理器定期运行以超时网络连接这很有用。

• InputRequirement：默认情况下所有处理器将允许用户为处悝器创建传入连接，但如果用户未创建传入连接则处理器仍然有效并可以安排运行。但是对于预期用作“源处理器”的处理器，这可能会使用户感到困惑并且用户可能会尝试将FlowFiles发送到该处理器，仅用于FlowFiles排队而不进行处理相反，如果处理器期望传入的FlowFiles但没有输入队列则处理器将被安排运行但不会执行任何工作，因为它将不会收到FlowFile这也会导致混淆。因此我们可以使用@InputRequirement注解，并为其提供一个值INPUT_REQUIREDINPUT_ALLOWED或INPUT_FORBIDDEN。这为框架提供了有关何时使处理器无效或者用户是否应该甚至能够将连接绘制到处理器的信息。例如如果使用注释处理器InputRequirement(Requirement.INPUT_FORBIDDEN)，则用户甚至无法创建与该处理器作为目标的连接

需要记住的一点是，NiFi提供了通用的数据处理功能数据可以是任何格式。处理器通常安排有多個线程开发人员对NiFi做出的一个常见错误是将FlowFile的所有内容缓冲到内存中。虽然有些情况需要这样做但应尽可能避免，除非众所周知数据嘚格式是什么例如，负责对XML文档执行XPath的处理器需要加载整个数据内容到内存中这通常是可以接受的，因为预计XML不会非常大但是，搜索特定字节序列的处理器可用于搜索数百千兆字节或更多的文件

不是将这些数据缓冲到内存中，而是建议在从内容存储库流式传输时评估数据（即扫描InputStream提供给回调的内容ProcessSession.read）。当然在这种情况下，我们不希望从Content Repository读取每个字节因此我们将使用BufferedInputStream或以某种方式缓冲一些少量數据，视情况而定

Services没有关系，因为它们没有直接集成到流中相反，它们由处理器报告任务和其他控制器服务使用。

控制器服务附带叻一个处理器没有的附加约束控制器服务必须包含扩展的接口ControllerService。然后实现只能通过其界面进行交互。例如处理器永远不会被赋予ControllerService的具体实现，因此必须仅通过扩展的接口引用服务ControllerService

这种约束主要是因为处理器可以存在于一个NiFi存档（NAR）中，而处理器所在的控制器服务的實现可以存在于不同的NAR中这是通过框架动态实现公开的接口来实现的，框架可以切换到适当的ClassLoader并在具体实现上调用所需的方法但是，為了使其工作处理器和控制器服务实现必须共享Controller Service接口的相同定义。因此这两个NAR必须依赖于包含Controller Service接口的NAR。有关更多信息请参阅。

使用此方法将提示用户提供应使用的SSL上下文服务。这是通过向用户提供下拉菜单来完成的他们可以从中选择已配置的任何SSLContextService配置，而不管实現如何

为了使用此服务，处理器可以使用以下代码：

到目前为止我们几乎没有提到如何向外界传达NiFi及其组件的表现。系统是否能够跟仩传入的数据速率系统还能处理多少？在一天的高峰时段与一天中最不忙的时间处理多少数据

为了回答这些问题以及更多问题，NiFi提供叻通过ReportingTask 界面向外部服务报告状态统计，指标和监控信息的功能ReportingTasks可以访问大量信息，以确定系统的运行方式

EventAccess通过ReportingContext公开的对象提供对该對象的访问ProcessGroupStatus，该对象公开有关过程组处理器，连接和其他组件在过去五分钟内处理的数据量的统计信息此外，EventAccess对象提供对存储在其中嘚ProvenanceEventRecords的访问ProvenanceEventRepository当从外部源接收数据，发送到外部服务从系统中删除，修改或根据所做出的决定路由时处理器发出这些源事件事件。

每个ProvenanceEvent嘟具有FlowFile的ID事件的类型，事件的创建时间以及组件访问FlowFile时与FlowFile关联的所有FlowFile属性以及与之关联的FlowFile属性。 FlowFile是事件描述的处理结果这为ReportingTasks提供了夶量信息，允许以多种不同方式生成报告以公开任何操作问题所需的指标和监控功能。

NiFi中有两个UI扩展点：

可以创建自定义UI以提供超出大哆数处理器设置中可用的标准属性/值表的配置选项具有自定义UI的处理器的示例是和。

可以创建内容查看器以扩展可在NiFi中查看的数据类型NiFi在lib目录中附带NAR，其中包含数据类型的内容查看器如csv，xmlavro，json（standard-nar）和图像类型如png，jpeg和gif（media-nar）

要将自定义UI添加到处理器：

1. 在处理器NAR中构建並捆绑WAR。

2. 将NAR放在lib目录中它将在NiFi启动时被发现。

3. 在处理器的“配置处理器”窗口中“属性”选项卡现在应该有一个Advanced按钮，该按钮将访问洎定义UI

1. 在处理器NAR中构建并捆绑WAR。

2. 将NAR放在lib目录中它将在NiFi启动时被发现。

3. 遇到匹配的内容类型时内容查看器将生成适当的视图。

一个很恏的例子是标准内容查看器的NAR布局：

标准内容查看器NAR布局

客户端/服务器操作模式来自于希望自动生成所需的TLS配置工件而无需在集中位置执荇该生成这简化了群集环境中的配置。由于我们不一定拥有运行生成逻辑或可信证书颁发机构的中心位置因此使用共享密钥来相互验證客户端和服务器。

tls-toolkit使用HMAC验证CA服务器的公钥和客户端发送的CSR来防止中间人攻击共享密钥（令牌）用作HMAC密钥。

1. 客户端生成包含CSR和HMAC的请求json有效负载其中令牌作为密钥，CSR的公钥指纹作为数据

2. 客户端连接到指定的https端口上的CA主机名，并验证CA证书的CN与主机名匹配（注意：因为此时峩们不信任CA这增加了无安全性，这只是一种方式如果可能的话提早错误）。

3. 服务器使用令牌作为密钥并使用CSR的公钥指纹作为数据从愙户端有效负载验证HMAC。这证明客户端知道共享密钥并且希望对具有该公钥的CSR进行签名。（注意：中间的人可以转发此功能但无法在不使HMAC无效的情况下更改CSR，从而无法实现目的）

4. 服务器对CSR进行签名，并发回包含证书的响应json有效负载和以令牌作为密钥的HMAC以及其公钥的指纹莋为数据

5. 客户端使用令牌作为密钥以及由TLS会话提供的证书公钥的指纹来验证响应HMAC。这验证了知道共享密钥的CA是我们通过TLS与之交谈的CA.

6. 客户端验证来自TLS会话的CA证书是否在有效负载中签署了证书

7. 客户端将生成的KeyPair添加到具有证书链的密钥库，并将CA证书从TLS连接添加到其信任库

8. 客戶端写出包含密钥库，信任库密码和有关交换的其他详细信息的配置json

测试将在更大的框架内使用的组件通常非常麻烦和棘手。通过NiFi我們努力使测试组件尽可能简单。为此我们创建了一个nifi-mock可与JUnit结合使用的模块，以提供对组件的广泛测试

模拟框架主要用于测试处理器，洇为它们是迄今为止最常开发的扩展点但是，该框架确实提供了测试Controller服务的能力

通常通过创建功能测试来验证组件行为，从而对组件進行测试这样做是因为处理器通常由少数辅助方法组成，但逻辑将主要包含在onTrigger方法中该TestRunner接口允许我们通过将更多的“原始”的对象，洳文件和字节数组到FlowFiles来测试处理器和控制器服务并处理创建ProcessSessions和所需的处理器ProcessContexts做它的工作，以及调用在必要的生命周期方法为了确保处理器在单元测试中的行为与在生产中的行为相同

处理器或控制器服务的大多数单元测试都是通过创建TestRunner 类的实例来开始的。为了向处理器添加必要的类您可以使用Maven依赖项：

我们TestRunner通过调用类的一个静态newTestRunner方法TestRunners（位于org.apache.nifi.util包中）来创建一个新的。这些方法为被测处理器提供了一个参数（可以是要测试的处理器的类也可以是处理器的实例），并允许设置处理器名称

在创建新的测试运行器之后，我们可以将任何控制器垺务添加到我们的处理器将需要的测试运行器以执行其工作我们通过调用该addControllerService方法并提供Controller Service的标识符和Controller Service的实例来完成此操作。

同样根据我們的期望，我们也可以调用assertValid()并assertNotValid()确保处理器的配置是否有效

在触发处理器运行之前，通常需要将FlowFiles排入队列以供处理器处理这可以通过使鼡该类的enqueue方法来实现TestRunner。该enqueue方法具有几个不同的覆盖并允许在的形式被添加的数据byte[]，InputStream或Path这些方法中的每一种还支持允许Map<String,

此外，还有一种enqueue方法可以获取FlowFile对象的var-args例如，这可用于获取处理器的输出然后将其提供给处理器的输入。

还有当我们想要触发处理器上运行，但不会觸发时间@OnUnscheduled和@OnStopped 生命周期事件例如，这有助于在这些事件发生之前检查处理器的状态这可以使用run(int, boolean)和传递false作为第二个参数来实现。但是在執行此操作后，调用@OnScheduled生命周期方法可能会导致问题因此，我们现在可以onTrigger再次运行而不会通过使用方法的run(int,boolean,boolean)版本run并false作为第三个参数传递来發生这些事件。

如果测试多个线程发生的行为很有用这也可以通过调用setThreadCount方法来实现 TestRunner。默认值为1个线程如果使用多个线程，请务必记住run调用TestRunner指定应触发处理器的次数，而不是每个线程应触发处理器的次数因此，如果线程计数设置为2但 run(1)被调用则只使用一个线程。

处理器运行完毕后单元测试通常需要验证FlowFiles是否符合预期要求。这可以使用TestRunnersassertAllFlowFilesTransferred和 assertTransferCount方法来实现前一种方法将关系和整数作为参数，以指示应该将哆少FlowFiles传输到该关系除非将此数量的FlowFiles转移到给定的关系或者任何FlowFile被转移到任何其他关系，否则该方法将无法通过单元测试该assertTransferCount方法仅验证FlowFile計数是给定关系的预期数量。

测试连接到远程资源的NiFi处理器时最大的问题之一是我们不希望从单元测试中实际连接到某些远程资源我们鈳以在单元测试中自己站起来一个简单的服务器并配置处理器与之通信，但是我们必须理解并实现特定于服务器的规范并且可能无法正確发送错误消息等我们想要测试。

通常这里采用的方法是在处理器中具有负责获得远程资源的连接或客户端的方法。我们通常将此方法標记为受保护在单元测试中，我们不是TestRunner通过调用TestRunners.newTestRunner(Class)和提供Processor类来创建而是在单元测试中创建处理器的子类并使用它：

这允许我们实现一个愙户端，该客户端模拟所有网络通信并返回我们想要测试的不同错误结果并确保我们的逻辑对于处理成功调用客户端是正确的。

除了测試框架提供的上述功能外TestRunner还提供了几种方便的方法来验证处理器的行为。提供了用于确保已清空处理器的输入队列的方法单元测试能夠获取ProcessContext，ProcessSessionFactoryProvenanceReporter以及TestRunner将使用的其他特定于框架的实体。该shutdown方法提供了测试注释为仅在关闭NiFi时运行的处理器方法的能力可以为使用自定义用户堺面的处理器设置注释数据。最后可以通过该setThreadCount(int)方法设置应该用于运行处理器的线程数。

当来自许多不同组织的软件都托管在同一环境中時Java ClassLoaders很快成为一个问题。如果多个组件依赖于同一个库但每个组件依赖于不同的版本则会出现许多问题，通常会导致出现意外行为或NoClassDefFoundError错誤为了防止这些问题成为问题，NiFi引入了NiFi存档或NAR的概念

NAR允许将多个组件及其依赖项一起打包到单个包中。然后NAR包与其他NAR包提供ClassLoader隔离。開发人员应始终将其NiFi组件部署为NAR包

为此，开发人员创建了一个新的Maven工件我们将其称为NAR工件。包装设定为nardependencies然后创建POM 的部分，以便NAR依赖於要包含在NAR中的所有NiFi组件

为了使用包装nar，我们必须使用该nifi-nar-maven-plugin模块通过将以下代码段添加到NAR的pom.xml中来实现此目的：

在Apache NiFi代码库中，这存在于NiFi根POMΦ其中所有其他NiFi工件（除了nifi-nar-maven-plugin本身）都继承，因此我们不需要将其包含在我们的任何其他工具中POM文件

Services就可以从任何处理器引用，无论它茬哪个NAR中

为了共享相同的定义，处理器的NAR和Controller服务实现的NAR必须作为父服务器服务定义的NAR示例层次结构可能如下所示：

组件开发人员可能唏望在运行时向组件的类路径添加其他资源。例如您可能希望将JDBC驱动程序的位置提供给与关系数据库交互的处理器，从而允许处理器使鼡任何驱动程序而不是尝试将驱动程序捆绑到NAR中。

在组件上设置这些属性后框架将标识所有dynamicallyModifiesClasspath设置为true的属性 。对于每个属性框架都尝試从属性的值中解析文件系统资源。该值可以是一个或多个目录或文件的逗号分隔列表其中跳过任何不存在的路径。如果资源表示目录则列出该目录，并且该目录中的所有文件都将单独添加到类路径中

通过将资源添加到始终首先检查的内部ClassLoader，将资源添加到实例ClassLoader只要這些属性的值发生更改，内部ClassLoader就会关闭并使用新资源重新创建

因为@RequiresInstanceClassLoading为组件的每个实例从NAR ClassLoader复制资源，所以明智地使用此功能如果创建了┅个组件的十个实例，则组件的NAR ClassLoader中的所有类都将加载到内存中十次当创建足够的组件实例时，这最终会显着增加内存占用

API与需要它的擴展捆绑在一起，即使未使用@RequiresInstanceClassLoading也会记录一条警告以帮助避免这种不良做法。

有时可能需要弃用组件每当发生这种情况时，开发人员可鉯使用@DeprecationNotice注释来指示组件已被弃用允许开发人员描述弃用的原因并建议替代组件。下面是一个如何执行此操作的示例：

如您所见替代方案可用于定义替代组件和数组，而classNames可用于通过字符串数组表示类似内容

我们总是很高兴能够从社区中获得贡献 - 尤其是来自新贡献者的贡獻！我们有兴趣接受代码的贡献，以及文档甚至可以作为图标或样式应用于应用程序的艺术作品

可用于查找标记为“初学者”的票证，戓者您可以深入了解创建处理器的任何票证处理器应该是独立的，不依赖于其他外部组件（Controller Services除外）因此它们为新的NiFi开发人员提供了良恏的起点。这使开发人员暴露于NiFi API并且是数据流系统中最具扩展性的部分。

可以通过创建补丁来提供贡献：

并将该补丁附加到故障单或苼成Pull请求。

有关贡献的更多详细信息请参阅的相关部分。

开发人员邮件列表（）受到密切监控我们倾向于快速响应。如果您有任何疑問请随时向我们发送电子邮件 - 我们随时为您提供帮助！不幸的是，电子邮件可能会在随机播放中丢失所以如果您发送电子邮件并且在┅两天内没有收到回复，那就是我们的错 - 不要担心打扰我们只需再次ping邮件列表。