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)
1、Nifi:基本認識
2、Nifi:基礎用法及頁面常識
3、Nifi:ExcuseXXXScript組件的使用(一)
4、Nifi:ExcuseXXXScript組件的使用(二)
5、Nifi:ExcuseXXXScript組件的使用(三)
6、Nifi:自定義處理器的開發