上一篇：Nifi組件腳本開發—ExecuteScript 使用指南（二）
Part 3 - 高級特征
本系列的前兩篇文章涵蓋了 flow file 的基本操作, 如讀寫屬性和內容, 以及使用"session" 變量 ( ProcessSession對象)獲取和轉移 flow files . ExecuteScript還有很多其他的能力，這里對一部分作簡要介紹。
動態屬性-Dynamic Properties
其中一個能力叫做 dynamic properties, 或者稱為用戶定義屬性. processor 的一些屬性可以由用戶設置 property name 和 value. 不是所有的processors 都支持和使用動態屬性, 在 ExecuteScript 將傳遞動態屬性作為變量，改變了引用 PropertyValue 對象，對應於property's value. 這里有兩個重要事需要了解：
- 因為 property 綁定為變量名, dynamic properties的命名規則必須滿足相應的編程語言的規范。 例如, Groovy 不支持 (.) 作為變量名字符, 像 "my.value" 引起processor處理失敗. 有效的可選項是 "myValue"
- PropertyValue 對象用於 (rather than a String representation of the value) 腳本執行多種操作，在轉換為String之前進行。如果property已知包含合法的值, 你可以調用 該變量的 getValue() 方法得到其字符串表示. 如果值包含 Expression Language，或者希望轉為除字符串外的其它值(如 'true' 對於Boolean 對象), 這里也提供了操作方法. 這些例子在下面的示例中演示, 假定我們有兩個屬性 'myProperty1' 和 'myProperty2'，像下面這樣被定義
獲取 dynamic property的值
需求：在腳本中得到 dynamic property(如, 配置參數)。
方法：使用變量的PropertyValue對象的getValue() 方法. 該方法返回其字符串代表 dynamic property. 注意，如果Expression Language包含在字符串中, getValue() 將不會對其求值(參加下一個方法實現求職功能)。
例子：
Groovy
def myValue1 = myProperty1.value
Jython
myValue1 = myProperty1.getValue()
Javascript
var myValue1 = myProperty1.getValue()
JRuby
myValue1 = myProperty1.getValue()
添加模塊
ExecuteScript 的另一個特征是具有添加外部模塊到 classpath 的能力, 這將允許使用大量的第三方庫、腳本等增強能力. 但是，每一個腳本引擎處理模塊的方法都是不一樣的, 因此需要分開討論。總體上說, 主要有兩種類型的模塊, Java libraries (JARs) 和 scripts (以在 ExecuteScript中的同一種語言編寫. 這里將討論和顯示不同 script engines 如何進行處理:
Groovy
Groovy script engine (至少在 ExecuteScript中) 不支持導入其他的 Groovy scripts, 但是允許 JARs 添加到 classpath. 因此，對於外部Groovy projects, 考慮編譯為bytecode，然后指向 classes 目錄或者包裝為 JAR.
當使用 Groovy, 這個 Module Directory 屬性設為 comma-separated 的文件列表 (JARs) 和 folders. 如果folder 被指定, ExecuteScript 將發現該目錄所有的 JARs 並添加進去. 這允許你添加第三方軟件，哪怕包含很多個 JARs.
Jython
Jython script engine (在 ExecuteScript) 目前僅支持導入純 Python 模塊, 不包含natively-compiled modules (如CPython)，如 numpy 或 scipy. 目前也暫不支持 JARs, 這在將來版本中也許會考慮. 在Module Directory property在執行前需要加載, 使用"import sys" 跟着 "sys.path.append" 對每一個指定的模塊位置進行加載.
如果 Python 已經安裝, 可以將所有的安裝好的純 Python modules 添加進來，通過將 site-packages 目錄加到Module Directory 屬性即可, 如：
/usr/local/lib/python2.7/site-packages
然后，你的腳本就能 import 各種軟件包了，如：
from user_agents import parse
Javascript
Javascript script engine (在 ExecuteScript), 允許同樣的 JARs/folders設置，與 Groovy engine一樣. 將查找JARs 以及指定的folder.
JRuby
JRuby script engine (在 ExecuteScript) 目前只允許單個的 JARs指定, 如果 folder 被指定，其中一定要有classes ( java compiler 需要能看見), 如果folder 包含 JARs將不會自動加入。目前, 沒有pure Ruby 模塊能被導入
狀態管理
NiFi (如0.5.0 ) 提供了為 Processors 和其他 NiFi 組件持久化一些信息從而實現組件的狀態管理功能. 例如, QueryDatabaseTable processor 保存對大數據集的跟蹤, 當下次運行時, 將只會獲取哪些比原來(存儲在 State Manager)更大的行的數據。
狀態管理的一個重要概念是Scope. NiFi 組件可以選擇存儲它的狀態在集群級別還是本地級別. 注意，在獨立的 NiFi 實例中, "cluster scope" 與 "local scope"是一樣的. 這個 scope 選擇的區別在於在一個數據流中，每個結點的處理器是否需要共享狀態信息. 如果集群中的實例不需要共享狀態，就使用local scope. 在 Java，這些選項作為一個 enum變量 Scope提供, 因此，當引用 Scope.CLUSTER 和 Scope.LOCAL, 就意味着是集群模式或本地模式.
為了探究ExecuteScript (語言獨立的例子如下)狀態管理的特征 , 你可以獲得 StateManager的引用，通過調用 ProcessContext的 getStateManager() 方法實現 (recall that each engine gets a variable named "context" with an instance of ProcessContext). 然后調用 StateManager 對象的下面方法:
- void setState(Map<String, String> state, Scope scope) - 在給定的scope更新組件狀態的值, 設置為給定的值. 注意，這個值是 Map 數據結構; 概念 "component state" 所有的 key/value鍵值對的 Map. 該 Map被一次全部更新，從而提供原子性.
- StateMap getState(Scope scope) - 返回組件在給定scope的當前狀態. 該方法永不會返回 null; 對於 StateMap 對象，如果 state沒有被設置, StateMap's 版本將是 -1, 而 map的值將是 empty. 經常，一個新的 Map<String,String> 被創建來存儲更新的值，然后setState()或 replace() 被調用.
- boolean replace(StateMap oldValue, Map<String, String> newValue, Scope scope) - 更新組件的狀態值 (在給定的 scope)為新的值，僅在當前值與給定的 oldValue一樣時執行. 如果 state 被更新為新的值, 返回true; 否則返回 false，如果state's value 不等於oldValue.
- void clear(Scope scope) - 從給定的scope下，清除組件狀態所有的鍵值
得到當前map的 key/value 對
需求：腳本從狀態管理器得到當前的 key/value 對，然后在 script 中使用(如更新等)。
方法：使用ProcessContext的getStateManager()方法, 然后從 StateManager調用 getStateMap() , 再 toMap() 轉換為Map<String,String>形式的key/value對. 注意，StateMap 也有 get(key) 方法去簡化獲得 value的方法, 但是不如 Map用的普遍。必須在 StateManager 一次性設置完畢。
例子：
Groovy
import org.apache.nifi.components.state.Scope
def oldMap = context.stateManager.getState(Scope.LOCAL).toMap()
Jython
from org.apache.nifi.components.state import Scope oldMap = context.stateManager.getState(Scope.LOCAL).toMap()
Javascript
var Scope = Java.type('org.apache.nifi.components.state.Scope'); var oldMap = context.stateManager.getState(Scope.LOCAL).toMap();
JRuby
java_import org.apache.nifi.components.state.Scope oldMap = context.stateManager.getState(Scope::LOCAL).toMap()
更新 key/value 映射的值對
需求：腳本希望通過新的包含key/value的映射值對來更新 state map。
方法：為了得到當前的 StateMap 對象, 再次用ProcessContext調用 getStateManager() 方法, 然后 StateManager調用getStateMap() . 例子中假定為新的 Map, 但是使用上面的配方 (通過 toMap() 方法), 你可以使用存在的值創建新的 Map, 然后用於更新想要的記錄. 注意，如果沒有當前map (i.e. the StateMap.getVersion() returns -1)，replace() 將不會工作, 因此例子中檢查並相應地調用 setState() 或 replace(). 當從ExecuteScript的新實例運行時，該StateMap 版本將會是 -1, 當單次執行后, 如果鼠標右鍵 ExecuteScript processor，然后選擇 View State, 將看到如下所示的信息：
例子：
Groovy
import org.apache.nifi.components.state.Scope
def stateManager = context.stateManager def stateMap = stateManager.getState(Scope.CLUSTER) def newMap = ['myKey1': 'myValue1'] if (stateMap.version == -1) { stateManager.setState(newMap, Scope.CLUSTER); } else { stateManager.replace(stateMap, newMap, Scope.CLUSTER); }
Jython
from org.apache.nifi.components.state import Scope
stateManager = context.stateManager
stateMap = stateManager.getState(Scope.CLUSTER)
newMap = {'myKey1': 'myValue1'}
if stateMap.version == -1:
stateManager.setState(newMap, Scope.CLUSTER)
else:
stateManager.replace(stateMap, newMap, Scope.CLUSTER)
Javascript
var Scope = Java.type('org.apache.nifi.components.state.Scope'); var stateManager = context.stateManager; var stateMap = stateManager.getState(Scope.CLUSTER); var newMap = {'myKey1': 'myValue1'}; if (stateMap.version == -1) { stateManager.setState(newMap, Scope.CLUSTER); } else { stateManager.replace(stateMap, newMap, Scope.CLUSTER); }
JRuby
java_import org.apache.nifi.components.state.Scope
stateManager = context.stateManager
stateMap = stateManager.getState(Scope::CLUSTER)
newMap = {'myKey1'=> 'myValue1'}
if stateMap.version == -1
stateManager.setState(newMap, Scope::CLUSTER)
else
stateManager.replace(stateMap, newMap, Scope::CLUSTER)
end
清空 state map
需求：清空 state map所有的e key/value 值。
方法：使用ProcessContext的getStateManager()方法, 然后調用StateManager的clear(scope)方法。
例子：
Groovy
import org.apache.nifi.components.state.Scope context.stateManager.clear(Scope.LOCAL)
Jython
from org.apache.nifi.components.state import Scopecontext.state Manager.clear(Scope.LOCAL)
Javascript
var Scope = Java.type('org.apache.nifi.components.state.Scope'); context.stateManager.clear(Scope.LOCAL);
JRuby
java_import org.apache.nifi.components.state.Scope context.stateManager.clear(Scope::LOCAL)
存取控制器服務
在 NiFi ARchive (NAR) 結構中, Controller Services-控制器服務被暴露為 interfaces, 在 API JAR中. 例如 , DistributedCacheClient 是一個從 ControllerService擴展來的接口, 位於 nifi-distributed-cache-client-service-api JAR中, 在 nifi-standard-services-api-nar NAR. 其他的 NARs 如果想要引用interfaces (去創建新的 client implementation, e.g.) 必須指定 nifi-standard-services-api-nar 作為父級 NAR, 然后在processor的子模塊提供 API JARs 的實例。
這是一些底層的細節，可能需要的以提升 Controller Services的使用, 這里提及主要是為了：
- 在 NiFi 1.0.0前, scripting NAR (包括 ExecuteScript 和 InvokeScriptedProcessor) 不需要指定nifi-standard-services-api-nar 作為父級. 這意味着只有明確的引用能被用於 ControllerServices 接口 (及其實現), 同樣的原因, 只有沒有要求其他不可用類的接口方法可以被使用. 這限制了 ExecuteScript 對Controller Services的使用
- NiFi 1.0.0, scripting processors 在nifi-standard-services-api-nar中存取 Controller Service interfaces (及其相關的classes) . 這包括DBCPService, DistributedMapCacheClient, DistributedSetCacheClient, HttpContextMap 和 SSLContextService. 但是我不相信nifi-standard-services-api-nar中其它的API 將會可用, 而且沒有定制化 ControllerService interfaces 將被識別
Processors 總是傾向於使用 Controller Service 實例創建 property (如PropertyDescriptor 對象) 並且調用 identifiesControllerService(class) . 當 UI component被渲染時, 將會發現所有的實現了期望接口的 Controller Services , component's ID 被使用, 友好顯示名稱被顯示給用戶。
對於ExecuteScript, 我們可以讓用戶選擇Controller Service 實例，通過讓他指定名稱或者 ID 來實現. 如果我們允許用戶指定name, 腳本將不得不執行一個查詢Controller Service實例列表去找到匹配名稱的元素。 這在上面的博客中提到了, 這里不再重復. 如果用戶輸入實例的ID, 然后 (在 NiFi 1.0.0) 將會更加容易滴匹配對象並存取，在下面將會看到. 這個例子將使用DistributedMapCacheClientService 實例為 "distMapClient", 連接到DistributedMapCacheServer 實例 (在標准的缺省配置下, localhost:4557), 這里 client instance 的ID為 93db6734-0159-1000-b46f-78a8af3b69ed：
在ExecuteScript 配置中, dynamic property被創建, 名為 "clientServiceId" 並且設為 93db6734-0159-1000-b46f-78a8af3b69ed:
然后我們使用clientServiceId.asControllerService(DistributedMapCacheClient), 這里參數是對DistributedMapCacheClient類對象的引用. 例如, 我有一個預先填充的緩存，字符串 key 'a' 設為字符串值 'hello'. 讓 Groovy script 使用 DistributedMapCacheServer進行工作。
一旦我們有了一個 DistributedMapCacheClient 實例, 然后就可以調用get(key, serializer, deserializer)去獲取值. 在這個例子中，因為keys 和 values 都是Strings, 我們只需要一個 Serializer<String> 和 Deserializer<String> 實例傳給 get() 方法. 該方法對於所有語言都是一樣的，通過 StreamCallback 實例的創建（在本系列文章的 Part 2）. 這個將從預先填充的服務器得到 key 'a' 的值，並且記錄值("Result = hello")。
得到property（存儲在 DistributedMapCacheServer）
需求：用戶發布值到 DistributedMapCacheServer (如配置數據)，然后使用腳本進行訪問。
方法：使用上面描述的方法，創建一個StringSerializer 和 StringDeserializer 對象, 然后通過ID得到DistributedMapCacheClientService 實例, 然后調用服務的 get() 方法. 記錄下結果到日志，方便后面查看。
例子：
Groovy
import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient
import org.apache.nifi.distributed.cache.client.Serializer
import org.apache.nifi.distributed.cache.client.Deserializer import java.nio.charset.StandardCharsets def StringSerializer = {value, out -> out.write(value.getBytes(StandardCharsets.UTF_8))} as Serializer<String> def StringDeserializer = { bytes -> new String(bytes) } as Deserializer<String> def myDistClient = clientServiceId.asControllerService(DistributedMapCacheClient) def result = myDistClient.get('a', StringSerializer, StringDeserializer) log.info("Result = $result")
Jython
from org.python.core.util import StringUtil from org.apache.nifi.distributed.cache.client import DistributedMapCacheClient, Serializer, Deserializer # Define a subclass of Serializer for use in the client's get() method class StringSerializer(Serializer): def __init__(self): pass def serialize(self, value, out): out.write(value) # Define a subclass of Deserializer for use in the client's get() method class StringDeserializer(Deserializer): def __init__(self): pass def deserialize(self, bytes): return StringUtil.fromBytes(bytes) myDistClient = clientServiceId.asControllerService(DistributedMapCacheClient) result = myDistClient.get('a', StringSerializer(), StringDeserializer()) log.info('Result = ' + str(result))
Javascript
var DistributedMapCacheClient =
Java.type('org.apache.nifi.distributed.cache.client.DistributedMapCacheClient'); var Serializer = Java.type('org.apache.nifi.distributed.cache.client.Serializer'); var Deserializer = Java.type('org.apache.nifi.distributed.cache.client.Deserializer'); var StandardCharsets = Java.type('java.nio.charset.StandardCharsets'); var StringSerializer = new Serializer(function(value, out) { out.write(value.getBytes(StandardCharsets.UTF_8)); }) var StringDeserializer = new Deserializer(function(arr) { // For some reason I had to build a string from the character codes in the "arr" array var s = ""; for(var i = 0; i < arr.length; i++) { s = s + String.fromCharCode(arr[i]); } return s; }) var myDistClient = clientServiceId.asControllerService(DistributedMapCacheClient.class); var result = myDistClient.get('a', StringSerializer, StringDeserializer); log.info("Result = "+ result);
JRuby
java_import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient
java_import org.apache.nifi.distributed.cache.client.Serializer
java_import org.apache.nifi.distributed.cache.client.Deserializer
java_import java.nio.charset.StandardCharsets
# Define a subclass of Serializer for use in the client's get() method
class StringSerializer include Serializer def serialize(value, out) out.write(value.to_java.getBytes(StandardCharsets::UTF_8)) end end # Define a subclass of Deserializer for use in the client's get() method class StringDeserializer include Deserializer def deserialize(bytes) bytes.to_s end end myDistClient = clientServiceId.asControllerService(DistributedMapCacheClient.java_class) result = myDistClient.get('a', StringSerializer.new, StringDeserializer.new) log.info('Result = ' + result)
原文地址：https://my.oschina.net/u/2306127/blog/858943
1、Nifi：基本認識
2、Nifi：基礎用法及頁面常識
3、Nifi：ExcuseXXXScript組件的使用（一）
4、Nifi：ExcuseXXXScript組件的使用（二）
5、Nifi：ExcuseXXXScript組件的使用（三）
6、Nifi：自定義處理器的開發