经过REST API更新 NiFi 数据流程

经过REST API更新 NiFi 数据流程

经过 NiFi 的REST API链接到服务并更新数据流程。

一、软件环境与预备知识

• NiFi 安装并在localhost下运行，参考: https://nifi.apache.org/download.html
• Groovy - 由于其 JSON builders 和 REST DSLs 很是好。若是使用Mac, 最简单的安装办法是： brew install groovy. 若是尚未安装Homebrew (Mac下的软件包管理器), 参考 http://brew.sh/
• 完整的脚本在GitHub: https://github.com/aperepel/nifi-rest-api-tutorial
• NiFi REST API 文档: https://nifi.apache.org/docs/nifi-docs/rest-api/index.html

二、建立测试流程

下面是一个测试流程：

准备这个测试流程：

• 添加 PutFile 组件到绘图面板。
• 重命名processor 为 Save File (right-click -> Configure -> Settings -> Name field)。后面将使用这个名称经过API查找 processor。
• 添加 GetHTTP processor，建立一个从 GetHTTP 到 ‘Save File’的链接。GetHTTP设置如今能够暂不设置，Save File processor 须要一个输入链接。
• 设置 Save File 属性以下 (这个设置后面能够经过编程进行修改)。
• 启动 Save File processor (这里不须要启动 GetHTTP)。

注意: 对于更复杂的流程，可使用模版来建立，参考: https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#templates

三、调用REST API

下一步, 咱们更新 Save File processor 使用不一样的目录 (/tmp/staging) 而且设置 “Create Missing Directories”为 “true”。

High-level script flow:

1. 查找 data flow中须要操做的组件。输入查找'Save File'，这与Web UI中的查找框使用的是同一个 API。

2.  确认这是惟一的 processor - 咱们但愿修改的事但愿的那一个处理节点。
3. 与framework的状态同步 - 取得最新版本的字段值, 将会用于下面的更新语句中。这是经典的乐观锁模式的实现。
4. 构建一个小的 JSON 文档，只包含状态改变。
5. 执行部分更新，经过 PUT 操做进行。
6. 重复 4-5，中止, 更新配置 (改变目录和目录属性)，启动 processor。

下面咱们直接执行这个脚本 (若是但愿保存、修改这个脚本，从github中clone/checkout到本地)：

groovy 
https://raw.githubusercontent.com/aperepel/nifi-rest-api-tutorial/master/reconfigure.groovy

能够看到输出信息，以下：

Looking up a component to update...
Found the component, id/group: c35f1bb7-5add-427f-864a-bdd23bb4ac7f/f1a2c4e8-b106-4877-97d9-9dbca868fc16

Preparing to update the flow state...
	Stopping the processor to apply changes...
	Updating processor...
	{
	"revision": {
	"clientId": "my awesome script",
	"version": 309
	},
	"processor": {
	"id": "c35f1bb7-5add-427f-864a-bdd23bb4ac7f",
	"config": {
	"properties": {
	"Directory": "/tmp/staging",
	"Create Missing Directories": "true"
	}
	}
	}
	}
	Updated ok.
	Bringing the updated processor back online...
	Ok

检查NiFi processor, 能够看到更新的目录和属性，而且建立了缺失的目录。除此以外, 每一步都被捕获下来并记录在 flow history中，以下：

当在UI中看见警告信息, 简单地点击Refresh连接刷新。在本文的后面将会介绍并发控制。

四、代码详解

首先, 拉取依赖项 https://github.com/jgritman/httpbuilder/wiki/RESTClient 。这个在 maven 仓库，构建时自动获取。

 @Grab(group='org.codehaus.groovy.modules.http-builder',
module='http-builder',
version='0.7.1')

这个库让咱们使用REST DSL，以下所示：

nifi.get(path: 'controller/search-results',query: [q: processorName])

nifi.put(path: "controller/process-groups/$processGroup/processors/$processorId",
    body: builder.toPrettyString(),requestContentType: JSON)

下一步, 使用Groovy的 JSON builder去构建一个JSON 文档，实现部分PUT更新。只须要指定但愿改变的属性，以下：

builder {

revision {
    clientId 'my awesome script'
    version resp.data.revision.version
    }

processor {
    id "$processorId"
    config {
        properties {
                'Directory' '/tmp/staging'
                'Create Missing Directories' 'true'
                }
            }
          }
}

这些dot-notation变量遍历JSON文档树。为了理解如何结构化这个返回结果，启动一个 GET 请求，将获取一个完整的 state 文档。

提示: UI 经过REST API来执行全部操做, 这是一个很是好的交互学习工具。注意，UI 对于PUT 和 POST (form) requests的操做是互换的, 因此选择在于那种用起来方便。这里咱们经过 PUT 和 JSON执行操做。

最后,  clientId 和 version 在下一节中进行介绍。

五、在 NiFi 中的乐观锁

下图展现了几本概念。

对于update操做，提供 clientId 是必须的，以免一致性问题(API 将返回 409 Conflict status 代码，若是开发者不知道这一点的话，将会引发困惑)。

controller/revision 返回用户的 最后修改流程的clientId，这不会老是你的 id。 最佳实践就是提供一个你的惟一ID以区分客户端。这实际上能够是任何格式的值, UUID 是框架在缺失时自动建立出来的缺省值。

英文：https://community.hortonworks.com/articles/3160/update-nifi-flow-on-the-fly-via-api.html

更多NiFi资源参考：http://www.javashuo.com/article/p-dkcrvjuz-nt.html