Hi Ravi, To distribute QueryDatabaseTable workload, I'd suggest using GenerateTableFetch instead. Because it can generate SQLs to query updated records. And those SQL FlowFiles can be distributed among NiFi nodes by RPG.
----- Following lines are just to share my thoughts on the topic for possible improvements. Ravi, Joe, how do you think? For QueryDatabaseTable, if there are multiple input tables, since it supports EL at table name, table names would be distributed among NiFi nodes to query each table from each NiFi node similar to List -> Fetch pattern. E.g. GenerateFlowFile (use 'Custom Text' to list target table names, only on primary node) -> SplitText (split per line) -> RPG -> QueryDatabaseTable GenerateFlowFile and SplitText part can be anything that can generate table names. QueryDatabaseTable also supports EL at 'Additional WHERE clause'. So if rows can be partitioned by particular column(s) and it's effectively useful for query (e.g. properly indexed) then it would be possible to distribute workload at record level by using the similar approach. However, above example will not work actually. Because NiFi manages cluster-wide processor state per processor id, the above example flow might updates the same state object from multiple NiFi nodes concurrently. It will result an optimistic lock error if that happens. To make it work, I'd suggest NiFi framework level improvements to support 'Custom State Key' for storing cluster-wide processor state. For example, with above example flow, QueryDatabaseTable would specify "table name + hash of additional where clause", then NiFi framework would store the state object with the provided custom state key and processor uuid. This way, those processors can distribute workload while keeping state cluster wide without competing updating the same state object. For GetHBase, currently it does not support EL at table name nor column family name. HBase scan can be distributed by executing partial scans per region (startRow and stopRow). Although GetXXXX processor do not take input FlowFiles usually, input FlowFile and EL support can be added to GetHBase. And add a processor to generate region info, GetHBaseRegions (start and end row ids), split and distribute it using RPG, then pass it to GetHBase. In this design, 'Custom State Key' for cluster-wide processor state will be needed, too. If it looks reasonable, I'd like to create JIRAs for possible future work. Thanks, Koji On Fri, Dec 22, 2017 at 3:58 AM, Ravi Papisetti (rpapiset) <[email protected]> wrote: > Thanks Joe for confirmation. > > I think, problem gets interesting when we start querying data from data bases > such as Oracle(QueryDatabaseTable) and Hbase (GetHbase). Are these also > expected to configure with “PrimaryNode”? RPG doesn’t seem to be recommended > to distribute huge content across cluster for data transfer. Any > recommendation how we should optimize and scale NiFi to transfer a huge hbase > (to HDFS or HBASE) or oracle table (to HDFS or HBASE) by taking advantage of > cluster. > > Thanks, > Ravi Papisetti > > On 21/12/17, 11:51 AM, "Joe Witt" <[email protected]> wrote: > > Ravi > > The pattern of List -> <distribut> -> Fetch -> Merge -> Put is common > and is the correct way to use it. > > ListFile can often be used across the cluster all at once and work out > fine since each node is often accessing a unique resource like a local > file system. However, in this case you're pointing them all at the > same source via a network share so they'll compete to pull data. > We're not sharing state between them as this coordination is extremely > costly. At least that is what it sounds like you're doing. So > running ListFile on primary only and then distributing the results is > correct. Before sending to HDFS you'll often want/need to merge > content first unless your objects are large enough to begin with. > > Thanks > Joe > > On Thu, Dec 21, 2017 at 12:17 PM, Ravi Papisetti (rpapiset) > <[email protected]> wrote: > > Hi, > > > > > > > > We are creating processgroups to transfer data from File system to > > HDFS/HBASE (separate process groups for each destination). > > > > > > > > Simple Example: FileSystem to HDFS process group: We have ListFile, > > FetchFile and PutHDFS process to transfer data from File system to HDFS. > > When “ListFile” processor’s execution mode is configured with “All > Nodes”, > > there is an error saying lease expired due to race condition. > > > > > > > > What I understood from exploration is, all nodes are trying to run > > “listFile” on target directory and all nodes in the cluster are trying > to > > transfer same files, when processing is complete by one of the node, > other > > node is throwing this exception message. > > > > > > > > In resolving that, we set “ListFIle” execution mode to run in > “PrimaryNode”, > > racing issue is resolved, but now all the files in target directory are > > processed by only one node in the cluster(no distribution). Data > transfer > > takes longer duration for large files. > > > > > > > > Further, we added “RPG” processor between “ListFile” and “FetchFile”, so > > that ListFile is executed by only one node and output of listfile is > > provided to RPG, further “FetchFile” and “PutHDFS” works in real cluster > > mode by processing each file by separate nodes in the cluster. > > > > > > > > This approach is applied for all type of file repositories (SFTP, NFS > and > > HDFS). It works well so far, but I would like to get experts opinion > here to > > make sure we are not missing any. > > > > > > > > Kind of surprised why NiFi is not working as expected with “ListFile” > > running in cluster mode, as a user I expect NiFi intelligently decide a > node > > to get list of files and FetchFile to work in cluster mode to > distribute the > > data transfer. > > > > > > > > Thanks, > > > > Ravi Papisetti > > > > > > > > > >
