Hi Joe,
Just completed by test with 100GB data (on a local RAID 5 disk on a single
server).
I was able to load 100GB data within 15 minutes(awesome!!) using below
flow. This throughput is enough to load 10TB data in a day with a single
and simple machine.
During the test, server disk I/O went up to 200MB/s.
ExecuteProcess(touch and mv to input dir) > ListFile > FetchFile (4
threads) > PutHDFS (4 threads)
My Next action is to incorporate my java code for column masking with a
custom processor.
I am now exploring on that. However, if you have any good reference on
custom processor(altering actual data) please let me know.
Thanks,
Obaid
On Mon, Jan 4, 2016 at 9:11 AM, obaidul karim <obaidc...@gmail.com> wrote:
> Hi Joe,
>
> Yes, symlink is another option I was thinking when I was trying to use
> getfile.
> Thanks for your insights, I will update you on this mail chain when my
> entire workflow completes. So that thus could be an reference for other :).
>
> -Obaid
>
> On Monday, January 4, 2016, Joe Witt <joe.w...@gmail.com> wrote:
>
>> Obaid,
>>
>> You make a great point.
>>
>> I agree we will ultimately need to do more to make that very valid
>> approach work easily. The downside is that puts the onus on NiFi to
>> keep track of a variety of potentially quite large state about the
>> directory. One way to avoid that expense is if NiFi can pull a copy
>> of then delete the source file. If you'd like to keep a copy around I
>> wonder if a good approach is to simply create a symlink to the
>> original file you want NiFi to pull but have the symlink in the NiFi
>> pickup directory. NiFi is then free to read and delete which means it
>> simply pulls whatever shows up in that directory and doesn't have to
>> keep state about filenames and checksums.
>>
>> I realize we still need to do what you're suggesting as well but
>> thought I'd run this by you.
>>
>> Joe
>>
>> On Sun, Jan 3, 2016 at 6:43 PM, obaidul karim <obaidc...@gmail.com>
>> wrote:
>> > Hi Joe,
>> >
>> > Condider a scenerio, where we need to feed some older files and we are
>> using
>> > "mv" to feed files to input directory( to reduce IO we may use "mv").
>> If we
>> > use "mv", last modified date will not changed. And this is very common
>> on a
>> > busy file collection system.
>> >
>> > However, I think I can still manage it by adding additional "touch"
>> before
>> > moving fole in the target directory.
>> >
>> > So, my suggestion is to add file selection criteria as an configurable
>> > option in listfile process on workflow. Options could be last modified
>> > date(as current one) unique file names, checksum etc.
>> >
>> > Thanks again man.
>> > -Obaid
>> >
>> >
>> > On Monday, January 4, 2016, Joe Witt <joe.w...@gmail.com> wrote:
>> >>
>> >> Hello Obaid,
>> >>
>> >> The default behavior of the ListFile processor is to keep track of the
>> >> last modified time of the files it lists. When you changed the name
>> >> of the file that doesn't change the last modified time as tracked by
>> >> the OS but when you altered content it does. Simply 'touch' on the
>> >> file would do it too.
>> >>
>> >> I believe we could observe the last modified time of the directory in
>> >> which the file lives to detect something like a rename. However, we'd
>> >> not know which file was renamed just that something was changed. So
>> >> it require keeping some potentially problematic state to deconflict or
>> >> requiring the user to have a duplicate detection process afterwards.
>> >>
>> >> So with that in mind is the current behavior sufficient for your case?
>> >>
>> >> Thanks
>> >> Joe
>> >>
>> >> On Sun, Jan 3, 2016 at 6:17 AM, obaidul karim <obaidc...@gmail.com>
>> wrote:
>> >> > Hi Joe,
>> >> >
>> >> > I am now exploring your solution.
>> >> > Starting with below flow:
>> >> >
>> >> > ListFIle > FetchFile > CompressContent > PutFile.
>> >> >
>> >> > Seems all fine. Except some confusion with how ListFile identifies
>> new
>> >> > files.
>> >> > In order to test, I renamed a already processed file and put in in
>> input
>> >> > folder and found that the file is not processing.
>> >> > Then I randomly changed the content of the file and it was
>> immediately
>> >> > processed.
>> >> >
>> >> > My question is what is the new file selection criteria for
>> "ListFile" ?
>> >> > Can
>> >> > I change it only to file name ?
>> >> >
>> >> > Thanks in advance.
>> >> >
>> >> > -Obaid
>> >> >
>> >> >
>> >> >
>> >> >
>> >> >
>> >> >
>> >> >
>> >> > On Fri, Jan 1, 2016 at 10:43 PM, Joe Witt <joe.w...@gmail.com>
>> wrote:
>> >> >>
>> >> >> Hello Obaid,
>> >> >>
>> >> >> At 6 TB/day and average size of 2-3GB per dataset you're looking at
>> a
>> >> >> sustained rate of 70+MB/s and a pretty low transaction rate. So
>> well
>> >> >> within a good range to work with on a single system.
>> >> >>
>> >> >> 'I's there any way to by pass writing flow files on disk or directly
>> >> >> pass those files to HDFS as it is ?"
>> >> >>
>> >> >> There is no way to bypass NiFi taking a copy of that data by
>> design.
>> >> >> NiFi is helping you formulate a graph of dataflow requirements from
>> a
>> >> >> given source(s) through given processing steps and ultimate driving
>> >> >> data into given destination systems. As a result it takes on the
>> >> >> challenge of handling transactionality of each interaction and the
>> >> >> buffering and backpressure to deal with the realities of different
>> >> >> production/consumption patterns.
>> >> >>
>> >> >> "If the files on the spool directory are compressed(zip/gzip), can
>> we
>> >> >> store files on HDFS as uncompressed ?"
>> >> >>
>> >> >> Certainly. Both of those formats (zip/gzip) are supported in NiFi
>> >> >> out of the box. You simply run the data through the proper process
>> >> >> prior to the PutHDFS process to unpack (zip) or decompress (gzip) as
>> >> >> needed.
>> >> >>
>> >> >> "2.a Can we use our existing java code for masking ? if yes then
>> how ?
>> >> >> 2.b For this Scenario we also want to bypass storing flow files on
>> >> >> disk. Can we do it on the fly, masking and storing on HDFS ?
>> >> >> 2.c If the source files are compressed (zip/gzip), is there any
>> issue
>> >> >> for masking here ?"
>> >> >>
>> >> >> You would build a custom NiFi processor that leverages your
>> existing
>> >> >> code. If your code is able to operate on an InputStream and writes
>> to
>> >> >> an OutputStream then it is very likely you'll be able to handle
>> >> >> arbitrarily large objects with zero negative impact to the JVM Heap
>> as
>> >> >> well. This is thanks to the fact that the data is present in NiFi's
>> >> >> repository with copy-on-write/pass-by-reference semantics and that
>> the
>> >> >> API is exposing those streams to your code in a transactional
>> manner.
>> >> >>
>> >> >> If you want the process of writing to HDFS to also do
>> decompression
>> >> >> and masking in one pass you'll need to extend/alter the PutHDFS
>> >> >> process to do that. It is probably best to implement the flow using
>> >> >> cohesive processors (grab files, decompress files, mask files, write
>> >> >> to hdfs). Given how the repository construct in NiFi works and
>> given
>> >> >> how caching in Linux works it is very possible you'll be quite
>> >> >> surprised by the throughput you'll see. Even then you can optimize
>> >> >> once you're sure you need to. The other thing to keep in mind here
>> is
>> >> >> that often a flow that starts out as specific as this turns into a
>> >> >> great place to tap the stream of data to feed some new system or new
>> >> >> algorithm with a different format or protocol. At that moment the
>> >> >> benefits become even more obvious.
>> >> >>
>> >> >> Regarding the Flume processes in NiFi and their memory usage. NiFi
>> >> >> offers a nice hosting mechanism for the Flume processes and brings
>> >> >> some of the benefits of NiFi's UI, provenance, repository concept.
>> >> >> However, we're still largely limited to the design assumptions one
>> >> >> gets when building a Flume process and that can be quite memory
>> >> >> limiting. We see what we have today as a great way to help people
>> >> >> transition their existing Flume flows into NiFi by leveraging their
>> >> >> existing code but would recommend working to phase the use of those
>> >> >> out in time so that you can take full benefit of what NiFi brings
>> over
>> >> >> Flume.
>> >> >>
>> >> >> Thanks
>> >> >> Joe
>> >> >>
>> >> >>
>> >> >> On Fri, Jan 1, 2016 at 4:18 AM, obaidul karim <obaidc...@gmail.com>
>> >> >> wrote:
>> >> >> > Hi,
>> >> >> >
>> >> >> > I am new in Nifi and exploring it as open source ETL tool.
>> >> >> >
>> >> >> > As per my understanding, flow files are stored on local disk and
>> it
>> >> >> > contains
>> >> >> > actual data.
>> >> >> > If above is true, lets consider a below scenario:
>> >> >> >
>> >> >> > Scenario 1:
>> >> >> > - In a spool directory we have terabytes(5-6TB/day) of files
>> coming
>> >> >> > from
>> >> >> > external sources
>> >> >> > - I want to push those files to HDFS as it is without any changes
>> >> >> >
>> >> >> > Scenario 2:
>> >> >> > - In a spool directory we have terabytes(5-6TB/day) of files
>> coming
>> >> >> > from
>> >> >> > external sources
>> >> >> > - I want to mask some of the sensitive columns
>> >> >> > - Then send one copy to HDFS and another copy to Kafka
>> >> >> >
>> >> >> > Question for Scenario 1:
>> >> >> > 1.a In that case those 5-6TB data will be again written on local
>> disk
>> >> >> > as
>> >> >> > flow files and will cause double I/O. Which eventually may cause
>> >> >> > slower
>> >> >> > performance due to I/O bottleneck.
>> >> >> > Is there any way to by pass writing flow files on disk or directly
>> >> >> > pass
>> >> >> > those files to HDFS as it is ?
>> >> >> > 1.b If the files on the spool directory are compressed(zip/gzip),
>> can
>> >> >> > we
>> >> >> > store files on HDFS as uncompressed ?
>> >> >> >
>> >> >> > Question for Scenario 2:
>> >> >> > 2.a Can we use our existing java code for masking ? if yes then
>> how ?
>> >> >> > 2.b For this Scenario we also want to bypass storing flow files on
>> >> >> > disk.
>> >> >> > Can
>> >> >> > we do it on the fly, masking and storing on HDFS ?
>> >> >> > 2.c If the source files are compressed (zip/gzip), is there any
>> issue
>> >> >> > for
>> >> >> > masking here ?
>> >> >> >
>> >> >> >
>> >> >> > In fact, I tried above using flume+flume interceptors. Everything
>> >> >> > working
>> >> >> > fine with smaller files. But when source files greater that 50MB
>> >> >> > flume
>> >> >> > chocks :(.
>> >> >> > So, I am exploring options in NiFi. Hope I will get some guideline
>> >> >> > from
>> >> >> > you
>> >> >> > guys.
>> >> >> >
>> >> >> >
>> >> >> > Thanks in advance.
>> >> >> > -Obaid
>> >> >
>> >> >
>>
>
