RouteOnAttribute

示例1

    使用Route做流程的分支，根据一个叫SO的栏位是否为空决定走不同的流程(左右的UpdateAttribute可以把它们想象成两个完全不同的处理流程来看)




    RouteOnAttribute，作用是根据Attribute的Bool值来决定是否进入该分支

• HasSO / NonSO：这是我自行定义的两个Route名称 (不是属性名称)，Value是一个表达式，若它的值=true，则Output会进该Route
• Nifi的表达式语法不怎么好写，官方的文档上有些东西并不支持 (不确定是不是Nifi版本缘故)。这里还是可以看得出来就只是判断一个叫SO的Attribute的值是否为空 --- 还记得有些Processor里面还有叫"Null Value Representation"的属性吧，若是那里配成"null"那这里也要与之匹配。
• 经过该Processor处理后，SO为空的会走右端逻辑，而不为空的则会走左段逻辑，同时它会加多一个叫"RouteOnAttribute.Route"的Attribute，内容即为Route名称
  

示例2：

    根据多个栏位是否有值(不为空)，让它们能够进入不同的Route，后面再根据Route名称去动态的取值，它的特点是Nifi流程并没有出现分支(Connection上是勾了所有的Route)，只是为不同的数据设上了一个变量名称



    RouteOnAttribute，借用分支名称在后面搞事，这里相当于是给资料加上了一个变量名称

• MO / MODELFAMILY / UPN / USN：这是定义的四个Route名称，判断条件都很简单，就只是不为空
• 值得注意的是这4个条件并非是互斥条件，比如有一笔资料它的MO、USN都不为空，那么就会同时进入两个Route进行输出，所以Output笔数会是2
• 下面这是进入UPN这个Route后的资料上的Attribute: "RouteOnAttribute.Route"
  
  

    UpdateAttribute，比较精彩的用法

• mqttTopic：这是定义的一个MQTT的Topic变量，它的内容是可变动的，会根据Route的不同产生不同的结果
• MO/MODELFAMILY/UPN/USN这4个条件有任一不为空，则会要求推送Topic： xxxx/【Type】/【Value】/yyyy
• 【Type】: 为"mo"、"modelfamily"、"upn"、"usn"这四个值之一
• 【Value】: 为MO/MODELFAMILY/UPN/USN这四个Attribute的取值 (即 ${MO} / ${MODELFAMILY} / ${UPN} / ${USN} 的值)，这里使用了双层${${"RouteOnAttribute.Route"}}的取值方法来实现动态的取用变量值

思路：

解决方案：

虽然与前一篇来说差异不大，但Nifi流程上却有很大不同，下面会详细描述为什么要这样做

Processor及其设定：

• SQL1：这才是这次花招的关键，在这里根据STATUS自行构建SQL语句
• Status=0：构建的就是Delete From xxx Where Key1=? and Key2=? 这样的删除语句
• Status!=0：构建的就是Replace Into xxx (Key1,Key2,col3,col4) Values ( ?, ?, ?, ?) 这样的Insert or Update语句
• 经此Processor处理后，资料落地所需的SQL就构建好了，后续的问题就是如何去绑定参数和执行

• Search Value：这里使用的是Default的(?s)(^.*$)，作用就是把原先的整份文件全部换掉
• Replacement Value：这里放的就是一个固定结构的JSON，可以看到里面的属性值都是使用的Attribute (它们的值来源于前面的EvaluateJsonPath从源JSON文件中的提取)
• 细心的朋友可以发现这里是与前一篇文章的最大不同，这里没有使用AttributeToJson去直接产生JSON文件，而使用的是更加笨拙的方式
• 前面的文章有提过，我们产生的Attribute以及AttributeToJson所生成JSON中各属性的顺序问题，结论是怎么搞它都不是我所想象到的顺序。但是ConvertJsonToSQL这个东东却很实在，它确确实实是按JSON中属性的顺序去生成的SQL以及参数名称(还记得参数名称sql.args.1.value中的这个顺序1么)，所以问题就来了：
  
• SQL由于必须要有Delete和Replace，所以它们的参数个数一定是不同的，而Delete压的参数又是我们的Key，所以就必须要保证ConvertJsonToSQL生成属性的顺序，这样我们才能够保证我们的两个Key一定会是sql.args.1和sql.args.2
• 换句话说，如果AttributesToJson若是能够保证JSON属性顺序的话，那就不用这么费劲

• Delete Attributes Expression：这里我根据Delete的条件(STATUS=0)去删除多余的SQL绑定参数
• 这里的写法比较死，我Hard-code删除掉大与2的其它所有参数(" * "是一个通配符，" | "是一个多条件的间隔符)，感觉上还有更好的写法
• 至此我们就可以保证绑定参数的数量与SQL语法参数个数一致 (不一致它死给你看)

• SQL Statement：前面用UpdateAttribute产生的SQL1参数，它会根据STATUS=0去判断是使用DELETE还是REPLACE语法
• 这个属性压上后，无论SQL1是不是为空，这个组件都不会再去管FlowFile的内容(空属性时是把FlowFile的内容当成SQL去执行的)

解决方案(仅新增、修改)：

    这个场景是最常见的，资料不会有被删除的状态，所有的更新就只有Insert Or Update这两种状态，先上实例的图 (两边的LogMessage是为了接收Fail，有感叹号是避免一起开启的时候它也被开启----这样failure的讯息就不会再卡在Connection中了)

思路：

Processor及其设定：

• Kafka Brokers：Kafka的Broker列表，多个Broker以逗号分隔，类似www.broker1.com:9193,www.broker2.com:9193这样的形式配置
• Topic Name：需要Consume的Kafka Topic名称
• Record Reader/Writer：关于Record所需要设定的Reader和Writer，要先行在Configure中设定，当然也要设定好Schema  
  
• Group ID：Consumer所要设定的ID，这个的设定要依Kafka的配置来，现在我们一般就只有单个的Partition，所以会要求每个Processor都设定有不同
• Offset Reset：需要设定为"earliest"，这样就会依GroupID没有收过的资料来进行收取，否则就只会收新推上去的资料。第一次玩的兄弟经常坑在GroupID和Offset Reset这两项上，若是收不到资料则有  可能就是GroupID没有换成新的(旧的已经收过一次就不会重新再收)，或者是Offset Reset = latest又没有新资料推上去~~~
  
• Max Poll Records和SCHEDULING中的Run Schedule：需要根据实际接收的速度来进行调整。经过观察发现Consume的速度超快，但整个Nifi Flow的速度会卡在其它需要做解析或读写DB的Processor外 (通常解析JSON会是前面的关卡)，所以任由Consumer的高速读取就会造成整个Nifi流程在后段被卡住。造成这个的主要原因其实就在于kafka处理的高速上，所以当有新换GroupID或新流程时，Kafka上积累的海量资料就会在一瞬间被接收下来，然后就是各种红 (其实红了也没事，它会自动向上推，让前一个Processor停止处理)。
  
• 若是常态下的资料推送量就已经超过了你的Nifi处理速度，那么就要考量使用多个线程处理或者是从源头的Kafka上就把资料分割开来  
• SCHEDULING的Cocurrent Tasks：这个Default=1，就是当前Processor需要开起来的线程数。但是这个设置需要当心，你需要仔细考量过你的资料流是否允许乱序 (多线程时当然不可能还能保证资料处理的顺序)，所以它是仅适用于不Care资料处理顺序的场景，例如每笔Key就只会有一笔资料，而且哪笔资料先收后收无所谓

将资料从Oracle推至MQTT，资料结果使用JSON格式
场景1：直接推送
场景2：仅当资料有变更时才推送

解决方案

Processor及其设定：

QueryDatabaseTable，作用是从DB中捞取资料，可以想象Nifi把它转成一个Select语句在执行

• Database Connection Pooling Service：在Configure中设定的数据库连接，可以在Processor Group中被共用
• Database Type：这个设定的是Oracle。其实它与数据库连接设定有点重叠，那边已经有指定是哪一种类型的DB，这里需要再指定，我想会不会是利用它来生成不同的Select查询语法?
• Table Name：表名，我这里使用的是View名称，View其实就已经对Table做出一些限定，可以挑选列及设定查询条件
• Maximum-value Columns：这个属性很重要，如果不设定则Nifi在查询的时候会捞取所有的资料，如果有设定某个列，则Nifi仅会捞取“新”资料
• 例如图上设定“BATCHID”这个列。第一次Nifi启动时捞取资料中最大BATCHID = 10，则下一次Nifi再次启动时就只会捞取BATCHID>10的资料，并且会自动记录下已经捞过的最大值
• 最大值保存在下面这里，Processor上右键选择“View State”
  
  
  
• 下图可以看到当前最新的值，点“Clear State”则可以将保留值清空（Nifi下次启动时就会捞取所有资料）
  
  
• 对于仅需要简单拉取资料的场景1来说，“Maximum-value Columns”置空即可；而对于仅在资料有更新时才要拉取的场景2来说，则需要设定并且在View中做出一些调整才可以达成
• 当有资料更新时，则被更新资料的BatchID会是更新的值，所以只要在View中虚拟BatchID列 = Max(BatchID)，就可以达成有资料更新才要拉取的效果

ConvertAvroToJSON，将Avro类型资料转换为JSON，这个可以不用改设定

PublishMQTT，将资料Publish到MQTT指定Topic

• Broker URI：MQTT的Broker地址
• Client ID：发布MQTT的Client端ID（注意不要多个Processor使用相同的Client ID，这样Processor容易被卡死）
• Topic：发布至MQTT的Topic名称
• Retain Message：MQTT的遗言属性，即是否保留推送的消息（若设为false，则仅有当前连上MQTT的客户端才能收到这笔消息）

問題 

   現在有遇到一個需求，是想直接利用TN5250來實現400端資料的下載：
   1. 僅通過TN5250來完成，不要使用其他的元件；
   2. 匯出資料格式為TXT檔案，應該是要能夠放到局域網路上；
   3. 匯出資料的動作需要做成排程方式(例如每天匯出)；
   4. 需求相當的急。

   使用TN5250NF可以直接抓400端資料到PC端，匯出成TXT/XLS等格式的文件。匯出功能實際由TN4002PC來完成，該功能使用起來非常方便。通過簡單設定連線/設定參數即可實現資料的下載。

   如果不考慮到需要做成排程方式來自動進行，則完全可以以人工方式來實現1-2。


思考

   要做直接利用程式的功能來實現自動化，首先應先弄清楚作業的方式是怎樣，先來看看手工是怎樣下載的。
   1. 點選TN5250上的菜單“傳送”-> “從主電腦接收檔案”即可打開功能畫面(圖1)：



   2. 需要配置參數(圖2)：

   >> 注意紅色框部分，輸出必須要設為“檔案”，而筆數=0則表示將該檔案中全部資料都DOWN，>0則表示最大下載筆數(如設為20則僅下載20筆資料)。其他部分根據需要自行配置。
   >> 檔案名稱處定義需要存放匯出檔案的名稱及路徑，如果需要在網路路徑上匯出文檔則必須先將網絡路徑映射為本地磁盤(通過“我的電腦”上右鍵“連線網路磁碟機”即可將一個網絡路徑映射成本地盤符)，然後再使用，否則會提示錯誤。

   3. 點選上圖的“開始接收(R)”即可以開始從400下載資料到PC，接收完成後會跳出訊息提示框(圖3)：


   另外，還可以利用圖2上的“檔案”菜單來實現對參數配置的保存以及載入(Default是存成.tt0檔案)。

   通過查看保存配置後生成的快捷方式，可以發現其快捷方式內容實際為(圖4)：

   >> 紅色部分即為我保存的配置檔--配置檔實際是一個INI文件，可使用記事本打開

   因此，使用“TN4002PC.EXE /f=配置檔名稱”這樣的形式即可從命令行實際對此功能的調用(先要將參數都配置好存在配置檔中)。OK，現在剩下就是要將這個命令行能夠被放在“計劃任務”中。

   通過手工測試可以發現有一些問題還待解決：
   1. 在匯出完成後一定會有訊息框跳出(圖3)，這個是相當於MSGBOX的窗口，它會將整個程式阻塞在那裏，如果不將它消掉(回車或空格)就會造成多次處理出錯。
   2. 配置參數時無法為匯出的TXT文檔指定一個動態名稱，因為是要自動將匯出資料放在一個固定路徑下，所以每回匯出的文檔名稱都應該有所不同，例如保存成帶日期格式。


解決方法

   因為是需要動態生成匯出文檔名以及消掉作業完成後的MSGBOX訊息框，所以是必須要有程序來協助解決，程序使用VBS來實現(先前有想到用BAT，但是無法解決MSGBOX的問題)。
   程式的處理方法：
   1. 利用VBS來動態生成TTO配置文件，只要每次修改掉保存的文件名稱就好，這樣就可以每次生成的TXT都有所不同。
         解析保存配置的TTO檔案，提取PC_FILENAME設定，在文件名後加上處理的日期和時間(路徑/文件類型保持不變)，最後再存成一個新的TTO配置檔。
   2. 利用VBS來執行CMD命令(命令格式見圖4)，完成對匯出功能的調用。
         為TN4002PC指定新生成的TTO配置檔，實現功能的調用。
   3. 在匯出完成後發送鍵盤指令關掉MSGBOX框。
         因為VBS中看似沒法能夠直接獲取TN4002PC的窗口，所以是添加處理--在程式處理過程中每間隔一段時間就去檢查生成的TXT檔案大小是否有變化，如果沒有變化則認定可能已匯出完成，卡在MSGBOX窗口，VBS程式就要激活MSGBOX窗口並發送鍵盤的空格。
   4. 在“計劃任務”中設置對VBS程序的調用。
         利用計劃任務精靈可以完成對VBS程序的調用，因為我這裏設計是要將配置的TTO文件名做為參數傳給VBS程序。而XP下試過幾次都無法能在計劃任務中直接將參數傳遞過去(2003卻可以)，所以就只有多做一只BAT(放置對VBS程式的傳參調用)放在計劃任務中。

BAT批處理程序(藍色即為VBS程序，紅色為保存的TTO配置檔)

VBS程序

VBS程序
Dim  ws,fso
Dim  ttoFile,tmpFile,sTmpfileName,sTxtfileName
Dim  line,sDate
' 當前的日期/時間
sDate  =   Year ( Now ())  &   Right ( " 0 " & Month ( Now ()), 2 )  &   Right ( " 0 " & Day ( Now ()), 2 )  &   Right ( " 0 "   &   Hour ( Now ()), 2 )  &   Right ( " 0 "   &   Minute ( Now ()), 2 )  &   Right ( " 0 "   &   Second ( Now ()), 2 )
Set  ws  =  WScript.CreateObject( " WScript.Shell " )
Set  fso  =   CreateObject ( " Scripting.FileSystemObject " )
' 新生成TTO配置檔,放在系統TEMP路徑中
sTmpfileName  =  fso.getspecialfolder( 2 )  &   " \ "   &  sDate  &   " .tto "
If  WScript.Arguments.Count > 0   Then
     Set  ttoFile  =  fso.openTextFile(WScript.Arguments( 0 ), 1 , False )
     Set  tmpFile  =  fso.openTextFile(sTmpfileName, 2 , True )
     While   Not  ttoFile.AtEndOfStream
        line  =  ttoFile.ReadLine()
         ' 查找到PC_FILENAME參數設定,並在文件名上加入日期/時間
         If   InStr ( 1 ,line, " PC_FILENAME= " ) = 1   Then
             If   InstrRev (line, " . " ) > 0   Then
                line  =   Mid (line, 1 , InstrRev (line, " . " ) - 1 )  &   " _ "   &  sDate  &   Right (line, Len (line) - InstrRev (line, " . " ) + 1 )
             Else
                line  =  line  &   " _ "   &  sDate
             End   If
            sTxtfileName  =   Mid (line, 13 )
         End   If
        tmpFile.WriteLine(line)
     Wend
    tmpFile.close
    ttoFile.close
     If   True   Then
     Dim  oExec,lastSize,newSize,blnFlag1
    blnFlag1  =   False
    lastSize  =   - 2
     ' 調用TN4002PC來完成匯出操作
     Set  oExec  =  ws.exec( " C:\TN5250NF\TN4002PC.EXE /f= "   &  sTmpfileName)
     Do   While  oExec.Status  =   0
        WScript.Sleep  5000
         If  blnFlag1  =   False   Then
             If  fso.FileExists(sTxtfileName)  Then
                newSize  =  fso.GetFile(sTxtfileName).Size
                 If  newSize  =  lastSize  Then
                    blnFlag1  =   True
                 Else
                    lastSize  =  newSize
                 End   If
             End   If
         Else
             ' 如果一段時間內匯出檔案大小無變化,則激活窗口並發送空格
            ws.AppActivate  " TN4002PC "  
            ws.SendKeys  "   "   
         End   If
     Loop
     End   If
End   If
Set  ws  =   Nothing   
WScript.quit

排程執行完後(每次都會有FDF--欄位定義和TXT--檔案內容兩個文件生成，多次RUN出文件名會不同):

下載
      VBS文件

-- 主檔(同一客戶會有多筆資料)
Create   Table  tbMain (
    aID  int   identity ( 1 , 1 ),
    pID  varchar ( 10 )  not   null ,     -- 客戶ID
    pName  nvarchar ( 20 )  not   null ,     -- 客戶姓名
    pDoing  varchar ( 20 )  not   null
);
-- 剔除客戶檔(僅存放客戶姓名)
Create   Table  tbForeclose (
    pName  nvarchar ( 20 )  not   null      -- 剔除客戶姓名
);
-- 客戶資料明細檔
Create   Table  tbDetail (
    pID  varchar ( 10 )  primary   key ,     -- 客戶ID(KEY)
    pSex  char ( 1 )  not   null          -- 客戶其他詳細資料
);

示例數據
SET  NOCOUNT  ON
Declare   @iRowCount   int
Declare   @pID   varchar ( 10 ), @pName   nvarchar ( 20 ), @pDoing   varchar ( 20 ), @pSex   char ( 1 )
Declare   @i   int , @j   int , @k   int
-- 生成tbMain記錄條數
Set   @iRowCount = 1000
Select   @i = 1
While   @i   <   @iRowCount
Begin
     -- 生成客戶ID
     Set   @j   =   1
     While   @j   >   0
     Begin
         Set   @pID   =   Left ( NewID (), 10 )
         Select   @j   =   Count ( * )  From  tbDetail  Where  pID = @pID
     End
     -- 生成客戶姓名
     Set   @k   =   2 + Round ( 3 * Rand (), 0 )
     Select   @j   =   1 , @pName = ''
     While   @j   <=   @k
     Begin
         Set   @pName   =   @pName   +   NChar ( 31000   +   Round ( 3000 * Rand (), 0 ))
         Set   @j   =   @j   +   1
     End
     -- 生成客戶性別
     If   @k   %   2   =   0
         Set   @pSex   =   ' T '
     Else
         Set   @pSex   =   ' F '
     -- 添加資料
     -- 主檔添加筆數(1-5筆)
     Set   @k   =   1   +   Round ( 5 * Rand (), 0 )
     Set   @j   =   1
     While   @j   <=   @k
     Begin
         Set   @i   =   @i   +   1
         -- 生成Doing欄位
         Set   @pDoing   =   Cast ( @j   As   Varchar ( 2 ))  +   ''' th Doing '
         Insert   Into  tbMain  Values  ( @pID , @pName , @pDoing )
         Set   @j   =   @j   +   1
     End
     -- 客戶資料明細檔
     Insert   Into  tbDetail  Values  ( @pID , @pSex )
End
-- 隨機向剔除客戶檔添加1/10的客戶姓名
Set   @k   =   Round ( @iRowCount   /   10 , 0 )
SET   ROWCOUNT    @k
Insert   Into  tbForeclose  Select  pName  From  tbMain  Group   By  pID,pName  Order   By   NewID ()
SET   ROWCOUNT   0
SET  NOCOUNT  OFF

嘗試及解決
      就此需求表面來看, 實現起來很是簡單, 只要一條SQL就好:

      但此需求在開發時卻有遇到一些限制條件, 不能這樣去實現.
      限制條件:
         1. tbMain,tbForeclose,tbDetail三表分別存在於不同的DB, 因DB所屬系統各不相同, 線上環境則會有可能布署在不同DBServer上.
         2. tbMain記錄筆數極大,索引效率低下.
         3. 三張TABLE是屬於其他的系統, 本次需求不允許進行UPDATE的操作.
      因此, 需要對應解決.

      對於限制條件1, 資料處理必須依3個步驟進行(見圖上A,B,C), 如圖1:
                        

      對於限制條件2, 主要卡在步驟A的超時上(其後都僅處理到少量資料), 實測時在本機的查詢分析器RUN步驟A的SQL就要10分鍾,符合的資料筆數為百萬左右.除非是將ADO連線設為不超時, 否則一定會查詢失敗.

      我首先的想法就是想要提高查詢效率. 
         . 經檢查tbMain發現pID列上有索引而pName上卻無, 所以將步驟A進行分解, 僅查詢符合的pID, 實測步驟A1可以在10分鍾以內完成, 見圖2.
                           
         . 嘗試減少步驟A1的處理筆數, 比如先依pID的前兩碼進行分組, 再進行多次查詢, 實測資料量最大的一組pID約在2-3分鍾, 單次資料量減少到十萬.
      通過測試的結果可以看到該問題僅通過改善/分解查詢已經不能實現解決. 

      那麽是否能夠將步驟A的筆數減少到一筆呢?
         . 因三檔都不允許進行UPDATE的操作, 在處理完TOP 1的一筆後因無法對處理過的pID進行標識, 無法獲取下一筆的資料.

      既然多種方法都不可行, 那麽就只有完全改變處理的流程, 通過多添加一張表(專為此開發而新增的TABLE)並將消耗最大的步驟A交由DB內部來完成, 最終解決方法步驟如下:
      1. 先手工新增臨時TABLE
      

--新增臨時檔
Create Table tbTemp (
    aID int identity(1,1),
    pID varchar(10)
);

      2. 如果是需要多次RUN的話, 則要先Truncate Table tbTemp, 然後通過Insert Into tbTemp Select....在DB內部完成資料的篩選, 由於在客戶端不需要撈一筆資料, 所以耗時非常的短, 達到秒級.
      3. 由於新增檔案是有KEY值(Identity), 所以只要拿到批次添加的最大/最小ID值即可通過循環來逐筆處理了.
      最終解決方案如圖3