Nifi組件腳本開發—ExecuteScript 使用指南（一）
Part 1 – 介紹 NiFi API 和 FlowFiles
ExecuteScript 是一個萬能的處理器，允許用戶使用程式語言定義自己的數據處理功能， 在每一次 ExecuteScript processor 觸發時被調用。下面的變數綁定到腳本環境，以提供腳本中訪問 NiFi 組件環境:
session: 是對processor的ProcessSession屬性的引用。session允許在 flow files 執行下面的操作： create(), putAttribute(), transfer(), 像 read() 和 write()一樣。
context: 是對 ProcessContext 的引用。可以用於檢索 processor 的屬性, 關係, Controller Services, 和 StateManager。
log: 是對ComponentLog的引用。用於 log 消息到 NiFi系統, 如 log.info(‘Hello world!’)。
REL_SUCCESS: 這是對 “success” relationship 的引用。這是從父類 (ExecuteScript)的靜態變數基礎來的, 但是一些引擎（如 Lua）不允許引用靜態成員, 只是一個為了方便的變數。
REL_FAILURE: 這是對 “failure” relationship 的引用。與 REL_SUCCESS 一樣, 這是從父類 (ExecuteScript)的靜態變數基礎來的, 但是一些引擎（如 Lua）不允許引用靜態成員, 只是一個為了方便的變數。
Dynamic Properties: 任何在ExecuteScript定義的動態屬性都作為變數集合到 PropertyValue 對象，對應於dynamic property。允許獲得property的 String 值 , 通過NiFi表達式進行求值，獲得相應的類型 (如 Boolean, 等等)。 因為動態屬性名稱成為腳本里的變數名, 你需要了解所選的腳本引擎的變數命名屬性。 例如, Groovy 不允許變數名中提供 (.) , 所以，如果 “my.property”作為動態屬性將會報錯。
與這些變數名的交互通過 NiFi Java API進行, 下面的每一個案例將討論相應的API調用。 下面的案例執行不同的函數操作 flow files, 如 reading/writing 屬性, 轉換為 relationship, logging, 等等。需要注意，這裡的例子只是一些片段。舉例來說, 如果使用session.get()從隊列中獲取 flow file , 必須轉換為 relationship 或者移除, 否則將會引發錯誤。程式碼片段應該是平面化的而且保持清晰，沒有容易引起混亂的程式碼，僅用於演示概念，從而讓工作簡單。
從incoming queue得到flow file
從session中獲得flow file
需求：從隊列中獲得輸入flow file，執行 ExecuteScript 並進行處理。
方法：使用 session對象的get(). 該方法返回FlowFile，是下一個最高優先順序的FlowFile用於處理. 如果沒有 FlowFile 用於處理, 該方法將返回 null. 注意, 如果一個持續的FlowFiles流進入processor，也可能會返回null. 這在多個並發任務處理時會發生，此時其他任務已獲得了 FlowFiles. 如果腳本要求有一個 FlowFile才能繼續處理, 如果是session.get()得到null應該立即返回。
例子：
Groovy：
flowFile = session.get() if(!flowFile) return
Jython：
flowFile = session.get() if (flowFile != None):
Javascript：
var flowFile = session.get();
if (flowFile != null) {
// All processing code goes here
}
JRuby：
flowFile = session.get() if flowFile != nil # All processing code goes here end
得到多個 flow files
需求：從queue(s)獲得多個flow files用於ExecuteScript處理
方法：使用session對象的get(maxResults) 方法. 該方法從工作隊列中返回最多 maxResults 個FlowFiles . 如果沒有 FlowFiles 可用, 一個空的list 將被返回 (而不是返回 null).。
例子：
Groovy：
flowFileList = session.get(100)
if(!flowFileList.isEmpty()) {
flowFileList.each { flowFile ->
// Process each FlowFile here
}
}
Jython：
flowFileList = session.get(100)
if not flowFileList.isEmpty():
for flowFile in flowFileList:
# Process each FlowFile here
Javascript：
flowFileList = session.get(100)
if(!flowFileList.isEmpty()) {
for each (var flowFile in flowFileList) {
// Process each FlowFile here
}
}
JRuby：
flowFileList = session.get(100)
if !(flowFileList.isEmpty())
flowFileList.each { |flowFile|
# Process each FlowFile here}
end
創建一個新的 flow files
創建新Flow Files
需求：創建一個新的 FlowFile 發送到下一步的 processor
方法：使用session的 create() 方法. 該方法返回 FlowFile 對象, 以用於後續的處理操作。
例子：
Groovy：
flowFile = session.create() // Additional processing here
Jython：
flowFile = session.create()
# Additional processing here
Javascript：
var flowFile = session.create();
// Additional processing here
JRuby：
flowFile = session.create()
# Additional processing here
從父級FlowFile創建新的 FlowFile
需求：從已有的 FlowFile 創建新的flow file 發送到下一步的 processor
方法：使用session的 create(parentFlowFile) 方法，該方法獲得父級 FlowFile 的引用，然後返回新的派生 FlowFile 對象。新創建的 FlowFile 除UUID之外將繼承父級的所有屬性，同時該方法將自動創建一個 起源 FORK 事件或 起源 JOIN 事件,在 ProcessSession被提交的時候，取決於FlowFiles 是否從同一個parent創建。
例子：
Groovy：
flowFile = session.get()
if(!flowFile) return
newFlowFile = session.create(flowFile)
// Additional processing here
Jython：
flowFile = session.get()
if (flowFile != None):
newFlowFile = session.create(flowFile)
# Additional processing here
Javascript：
var flowFile = session.get();
if (flowFile != null) {
var newFlowFile = session.create(flowFile);
// Additional processing here
}
JRuby：
flowFile = session.get()
if flowFile != nil
newFlowFile = session.create(flowFile)
# Additional processing here
end
flow file 的attributes操作
從 flow file 得到屬性
需求：獲得flow file 的屬性。
方法：使用FlowFile對象getAttribute(attributeKey) 。 該方法對於給定的attributeKey返回一個字元串值 , 如果沒有找到相應的key就返回null. 下面的例子演示返回FlowFile的 “filename” 屬性。
例子：
Groovy：
flowFile = session.get()
if(!flowFile) return
myAttr = flowFile.getAttribute('filename')
Jython：
flowFile = session.get()
if (flowFile != None):
myAttr = flowFile.getAttribute('filename')
# implicit return at the end
Javascript：
var flowFile = session.get()
if (flowFile != null) {
var myAttr = flowFile.getAttribute('filename')
}
JRuby：
flowFile = session.get()
if flowFile != nil
myAttr = flowFile.getAttribute('filename')
end
從 flow file得到所有的屬性
需求：從flow file得到所有的屬性。
方法：使用FlowFile對象的getAttributes() 方法。 該方法返回 Map 數據結構，由字元串的 keys 和 values組成, 代表一個FlowFile的屬性的 key/value 值對。 下面的顯示如何遞歸顯示FlowFile的所有屬性的Map的值。
例子：
Groovy：
flowFile = session.get()
if(!flowFile) return
flowFile.getAttributes().each { key,value ->
// Do something with the key/value pair
}
Jython：
flowFile = session.get()
if (flowFile != None):
for key,value in flowFile.getAttributes().iteritems():
# Do something with key and/or value
# implicit return at the end
Javascript：
var flowFile = session.get()
if (flowFile != null) {
var attrs = flowFile.getAttributes();
for each (var attrKey in attrs.keySet()) {
// Do something with attrKey (the key) and/or attrs[attrKey] (the value)
}
}
JRuby：
flowFile = session.get()
if flowFile != nil
flowFile.getAttributes().each
{ |key,value|
# Do something with key and/or value
}
end
添加屬性到 flow file
需求：在已有的 flow file 上添加自己的屬性。
方法：使用session對象的 putAttribute(flowFile, attributeKey, attributeValue) 方法。 該方法更新給定的 FlowFile’s 屬性，使用給出的 key/value 對來進行。
注意：對象的 “uuid” 屬性是固定的，並且不能修改; 如果key被命名為 “uuid”, 將被忽略.
這裡的FlowFile 對象是不可改變的; 這意味著，如果通過API更新了 FlowFile 的屬性 (或其它的改變了) , 你將得到一個新版的FlowFile的新的引用。當轉換FlowFiles到relationships時這是非常重要的。你必須保持對FlowFile的最新版本的引用, 你必須轉換或者移除所有的FlowFiles的最後版本, 否則執行時將會得到錯誤資訊。經常情況下, 該用於存儲 FlowFile 引用變數將會被最後返回的版本覆蓋 (中間的 FlowFile 應用將會被自動拋棄). 在這個例子中，你可以看到當添加屬性時重用flowFile引用的技術。注意到當前的flowFile引用被傳遞給putAttribute() 方法. 這個結果FlowFile具有命名為 'myAttr'值為 'myValue'的屬性。如果你有一個對象，可以序列化為String. 最終, 請注意如果你添加了多個屬性, 最好創建一個Map，然後使用 putAllAttributes() 方法來進行賦值。
例子：
Groovy：
flowFile = session.get()
if(!flowFile) return
flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
Jython：
flowFile = session.get()
if (flowFile != None):
flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
# implicit return at the end
Javascript：
var flowFile = session.get();
if (flowFile != null) {
flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
}
JRuby：
flowFile = session.get()
if flowFile != nil
flowFile = session.putAttribute(flowFile, 'myAttr', 'myValue')
end
添加多個屬性到一個flow file
需求：向 flow file 添加多個自定義屬性。。
方法：使用 session對象的putAllAttributes(flowFile, attributeMap) 方法。該方法更新給定的FlowFile’s 屬性，以 key/value 對的方式存儲在Map中返回。
注意：對象的 “uuid” 屬性是固定的，並且不能修改; 如果key被命名為 “uuid”, 將被忽略.
該技術創建了一個 Map (aka dictionary in Jython, hash in JRuby) 用於更新，然後調用putAllAttributes() 。這比對putAttribute() 對每一個 key/value 遍歷效率更高, 這將導致對每一個屬性調用時 FlowFile 都需要創建一個副本 (查看上面 FlowFile 不變性的討論)。下面例子中的Map包含兩個條目： myAttr1 和 myAttr2, 設為 '1' 並且第二個為 String (附著到方法簽名，對key和value都要求 String)。 注意到session.transfer() 在這裡並未指定 (因此下面的程式碼片段並不能工作), 查看下面的方法。
例子：
Groovy：
attrMap = ['myAttr1': '1', 'myAttr2': Integer.toString(2)] flowFile = session.get() if(!flowFile) return flowFile = session.putAllAttributes(flowFile, attrMap)
Jython：
attrMap = {'myAttr1':'1', 'myAttr2':str(2)}
flowFile = session.get()
if (flowFile != None):
flowFile = session.putAllAttributes(flowFile, attrMap)
# implicit return at the end
Javascript：
var number2 = 2;
var attrMap = {'myAttr1':'1', 'myAttr2': number2.toString()}
var flowFile = session.get()
if (flowFile != null) {
flowFile = session.putAllAttributes(flowFile, attrMap)
}
JRuby：
attrMap = {'myAttr1' => '1', 'myAttr2' => 2.to_s}
flowFile = session.get()
if flowFile != nil
flowFile = session.putAllAttributes(flowFile, attrMap)
end
轉換 flow file
轉移一個flow file 到 relationship
需求：在處理完flow file (new or incoming)之後, 你希望將flow file轉移到 relationship (“success” or “failure”). 在這個簡單的例子中，讓我們假定有一個變數叫做 “errorOccurred”， 用於指示在哪種 relationship下 FlowFile 將被轉移。
方法：使用session對象的transfer(flowFile, relationship) 方法。基於給定的relationship，該方法將給定的FlowFile發送到適合的目標處理器隊列。如果relationship通向不止一個目標，FlowFile的狀態將被複制 ，從而每一個目標都將收到一個 FlowFile的拷貝，因此也將具有唯一的標識符UUID。
注意：最後，ExecuteScript將執行session.commit() 以進行操作的提交。你不需要在腳本內部執行session.commit() 來執行提交操作。
例子：
Groovy：
flowFile = session.get()
if(!flowFile) return
// Processing occurs here
if(errorOccurred) {
session.transfer(flowFile, REL_FAILURE)
}
else {
session.transfer(flowFile, REL_SUCCESS)
}
Jython：
flowFile = session.get()
if (flowFile != None):
# All processing code starts at this indent
if errorOccurred:
session.transfer(flowFile, REL_FAILURE)
else:
session.transfer(flowFile, REL_SUCCESS)
# implicit return at the end
Javascript：
var flowFile = session.get();
if (flowFile != null) {
// All processing code goes here
if(errorOccurred) {
session.transfer(flowFile, REL_FAILURE)
}
else {
session.transfer(flowFile, REL_SUCCESS)
}
}
JRuby：
flowFile = session.get()
if flowFile != nil
# All processing code goes here
if errorOccurred
session.transfer(flowFile, REL_FAILURE)
else
session.transfer(flowFile, REL_SUCCESS)
end
end
日誌 Logging
發送消息到 log並制定日誌級別
需求：希望報告一些事件、消息並通過日誌框架寫入。
方法： 使用 log 的方法(), trace(), debug(), info(), 或 error() 完成。這些方法可以是單個的字元串或者字元串數組對象, 或字元串後面跟著Throwable的對象數組。第一個用於簡單消息. 第二個用於一些動態對象（值）的log。在消息字元串中使用 “{}” 進行引用。這些用於對對象數組進行求值，當消息讀到 “Found these things: {} {} {}” 並且 Object array 是 [‘Hello’,1,true], 那麼logged 消息將是 “Found these things: Hello 1 true”，第三種logging方法帶一個 Throwable 參數, 這在例外被捕捉到並且希望日誌記錄時使用。
例子：
Groovy：
log.info('Found these things: {} {} {}', ['Hello',1,true] as Object[])
Jython：
from java.lang import Object
from jarray import array
objArray = ['Hello',1,True]
javaArray = array(objArray, Object)
log.info('Found these things: {} {} {}', javaArray)
Javascript：
var ObjectArrayType = Java.type("java.lang.Object[]");
var objArray = new ObjectArrayType(3);
objArray[0] = 'Hello';
objArray[1] = 1;
objArray[2] = true;
log.info('Found these things: {} {} {}', objArray)
JRuby：
log.info('Found these things: {} {} {}', ['Hello',1,true].to_java)
