You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-user@hadoop.apache.org by Keith Wiley <kw...@keithwiley.com> on 2011/02/03 18:16:16 UTC
Streaming data locality
I've seen this asked before, but haven't seen a response yet.
If the input to a streaming job is not actual data splits but simple HDFS file names which are then read by the mappers, then how can data locality be achieved.
Likewise, is there any easier way to make those files accessible other than using the -cacheFile flag? That requires building a very very long hadoop command (100s of files potentially). I'm worried about overstepping some command-line length limit...plus it would be nice to do this programatically, say with the DistributedCache.addCacheFile() command, but that requires writing your own driver, which I don't see how to do with streaming.
Thoughts?
Thanks.
________________________________________________________________________________
Keith Wiley kwiley@keithwiley.com www.keithwiley.com
"I used to be with it, but then they changed what it was. Now, what I'm with
isn't it, and what's it seems weird and scary to me."
-- Abe (Grandpa) Simpson
________________________________________________________________________________
Re: Streaming data locality
Posted by Keith Wiley <kw...@keithwiley.com>.
On Feb 3, 2011, at 6:25 PM, Allen Wittenauer wrote:
>
> On Feb 3, 2011, at 9:16 AM, Keith Wiley wrote:
>
>> I've seen this asked before, but haven't seen a response yet.
>>
>> If the input to a streaming job is not actual data splits but simple HDFS file names which are then read by the mappers, then how can data locality be achieved.
>
> If I understand your question, the method of processing doesn't matter. The JobTracker places tasks based on input locality. So if you are providing the names of the file you want as input as -input, then the JT will use the locations of those blocks.
Let's see here. My streaming job has a single -input flag which points to a text file containing HDFS paths. Each line contains one TAB. Are you saying that if the key (or the value) on either side of that TAB is an HDFS file path then that record will be assigned to a task in a data local manner? Which is it that determines this locality, the key or the value? (Must be the key, right?)
> (Remember: streaming.jar is basically a big wrapper around the Java methods and the parameters you pass to it are essentially the same as you'd provide to a "real" Java app.)
>
> Or are you saying your -input is a list of other files to read? In the case, there is no locality. But again, streaming or otherwise makes no real difference.
Yes, basically. The input is a list of HDFS file paths to be read and processed on a an individual basis.
>> Likewise, is there any easier way to make those files accessible other than using the -cacheFile flag?
>> That requires building a very very long hadoop command (100s of files potentially). I'm worried about overstepping some command-line length limit...plus it would be nice to do this programatically, say with the DistributedCache.addCacheFile() command, but that requires writing your own driver, which I don't see how to do with streaming.
>>
>> Thoughts?
>
> I think you need to give a more concrete example of what you are doing. -cache is used for sending files with your job and should have no bearing on what your input is to your job. Something tells me that you've cooked something up that is overly complex. :D
Good point, I'll write a better description of this later. Thanks for the advice.
________________________________________________________________________________
Keith Wiley kwiley@keithwiley.com keithwiley.com music.keithwiley.com
"I used to be with it, but then they changed what it was. Now, what I'm with
isn't it, and what's it seems weird and scary to me."
-- Abe (Grandpa) Simpson
________________________________________________________________________________
Re: Streaming data locality
Posted by Allen Wittenauer <aw...@linkedin.com>.
On Feb 3, 2011, at 11:03 PM, Keith Wiley wrote:
> Our data consists of a few million binary (image) files, only about 6MB each uncompressed, 2.5MB gzipped. A typical job does not need to touch every image, rather only a tiny subset of the images is required. In this sense our job is unlike a canonical Hadoop job which combs an entire database. As such our current implementation (not streaming, but "normal" Hadoop) first uses a SQL query against a nonHadoop database to find the names of the relevant images for a given job, then sets up input splits for those individual files and assigns them as input to the Hadoop job.
Why can't you just process the entire dataset, skipping the records you don't care about? In other words, your input to your streaming job would be all the files. But you could send a list of the actual images you care about as a dcache file. Then you'd get locality while also only processing the ones you need.
[This is a very common pattern, BTW.]
> ....
> Which brings me to streaming. Serendipitously, our system seems to work considerably better if I don't use Java mappers and JNI, but rather using streaming to run a small native program that then coordinates the calls into the native library (further evidence that the problem lies somewhere in JNI's functionality). I am not sure how to arrange our input data for streaming however. It is binary of course, which has been a long-standing issue (I believe) with Hadoop streams since the assumption is that the data is 8-bit clean with endline record terminators.
I'm fairly certain this is actually fixed in either 0.21 or trunk. So this long standing issue will soon go away. :D
> Furthermore, I'm not really sure how I would indicate which files are the input for a given job since I don't know how to write a driver program for streaming, I only know how to use the streaming jar directly from the command line (so there is no driver of my choosing, I can't control the driver at all, its the "streaming" driver and all I can do is give it my mapper and reducer native programs to run. You see my problem? With streaming I have no driver, just the mappers and reducers, but with a Java Hadoop program, I have all three and I can do useful preprocessing in the driver, like setting up the input paths to the subset of images comprising the input dataset for a given job. This is quite perplexing for me. How do I use streams effectively if I can't write my own driver?
Again, I think a way around this dilemma is to basically do record skipping. It is understood that streaming has a lot of limitations. The fact that it works at all is pretty amazing. :)
Re: Streaming data locality
Posted by Keith Wiley <kw...@keithwiley.com>.
On Feb 3, 2011, at 6:25 PM, Allen Wittenauer wrote:
> I think you need to give a more concrete example of what you are doing. -cache is used for sending files with your job and should have no bearing on what your input is to your job. Something tells me that you've cooked something up that is overly complex. :D
Fair enough. Here's my setup.
Our data consists of a few million binary (image) files, only about 6MB each uncompressed, 2.5MB gzipped. A typical job does not need to touch every image, rather only a tiny subset of the images is required. In this sense our job is unlike a canonical Hadoop job which combs an entire database. As such our current implementation (not streaming, but "normal" Hadoop) first uses a SQL query against a nonHadoop database to find the names of the relevant images for a given job, then sets up input splits for those individual files and assigns them as input to the Hadoop job.
Side note: since the images are so small and we use so many at a time, they are packed into sequence files. Building the input splits is a very precise piece of code since it must splice out the necessary image files from their locations in the sequence files as eventual Hadoop input splits.
The mappers receive byte arrays as input from the record reader (in the form of a WritableComparable class that contains the byte array).
Our computational heavy lifting resides in a very large and complex set of native libraries (C++) which our Java mappers call using JNI, passing the byte arrays (the image files slurped into memory) through JNI to a small native .so file which then coordinates the processing with the underlying processing library. The native routines return byte arrays through JNI that represent the converted images. The Java mappers pass the byte arrays on to the reducer stage for final processing. In our current design, the reducer's computational complexity is simple enough to be implemented and performed entirely in Java, sans JNI or native code.
This system works fine at an organizational level, but unfortunately we have been plagued by extremely insidious segfaults and memory-allocation exceptions deep in the native library, bugs which absolutely never occur in a nonHadoop or nonJava environment when the library is accessed solo from a native C++ program (or from python scripts).
I have determined that the problem is not an incompatibility between the native library and Hadoop since a nonHadoop Java program, using the same JNI interface to access our native routines, exhibits the same errors. This is some fundamental incompatibility between Java/JVM/JNI and our native library. Like I said, the library is very complicated. It involves extremely fancy template programming and I suspect JNI is simply not properly coordinating memory between the two environments. I've valgrinded it, I've looked at the core dumps, I haven't found an underlying cause yet, much less fixed the problem.
Which brings me to streaming. Serendipitously, our system seems to work considerably better if I don't use Java mappers and JNI, but rather using streaming to run a small native program that then coordinates the calls into the native library (further evidence that the problem lies somewhere in JNI's functionality). I am not sure how to arrange our input data for streaming however. It is binary of course, which has been a long-standing issue (I believe) with Hadoop streams since the assumption is that the data is 8-bit clean with endline record terminators. Furthermore, I'm not really sure how I would indicate which files are the input for a given job since I don't know how to write a driver program for streaming, I only know how to use the streaming jar directly from the command line (so there is no driver of my choosing, I can't control the driver at all, its the "streaming" driver and all I can do is give it my mapper and reducer native programs to run. You see my problem? With streaming I have no driver, just the mappers and reducers, but with a Java Hadoop program, I have all three and I can do useful preprocessing in the driver, like setting up the input paths to the subset of images comprising the input dataset for a given job. This is quite perplexing for me. How do I use streams effectively if I can't write my own driver?
Anyway, what I've come up with is to create a text file that lists the input files relevant to a job. That one text file is the streaming input, each line of which is an input record. Each line of the file is a properly formatted streaming key/val pair (meaning a TAB separates the key and the value). Interestingly, I don't need the key/value pairs per se, I only need the input filenames. So basically I don't use the key very much and store the input files names in the value (meaning right after the TAB on each line of the input file). My native program obviously receives a set of these lines from Hadoop and opens each file and operates on it. To do this, the file must be accessible the the native code, and the only way I know to do that is for the file to be in the distributed cache (which is aliased to the current working directory of the native program), thus my previously described method of putting symbolic links to all the input files in the distributed cache using the -cacheFile flag. That way the native program can open them when it finds their names in the streaming input records.
So that's what I've got going on. If you have a better design in mind, I would love to hear about it. I admit, this all feels rather contrived, but I simply don't know how else to do it. The documentation is rather thin on these topics so I'm kind of making it up as I go along. For example, I have not found a single code example anywhere demonstrating how to use the streaming API with a user-defined driver. These are examples of this for Java-only Hadoop programs so that made it easy for me to learn how to do that, but for streaming, no such luck -- I'm not even sure such a thing is possible with streaming. Furthermore, I remain confounded how to set up or otherwise specify a subset of files on HDFS as input records to streaming. I could give it the root directory of my entire database and it would happily feed all of the images to my job, but like I said, I only need a tiny subset for any one job. My driver would usually handle this on a per-file basis, but I don't have one in streaming. Instead, I give it a single file as streaming input which contains a list of relevant files in HDFS for processing a given job.
Thank you very much.
Cheers!
________________________________________________________________________________
Keith Wiley kwiley@keithwiley.com keithwiley.com music.keithwiley.com
"The easy confidence with which I know another man's religion is folly teaches
me to suspect that my own is also."
-- Mark Twain
________________________________________________________________________________
Re: Streaming data locality
Posted by Allen Wittenauer <aw...@linkedin.com>.
On Feb 3, 2011, at 9:16 AM, Keith Wiley wrote:
> I've seen this asked before, but haven't seen a response yet.
>
> If the input to a streaming job is not actual data splits but simple HDFS file names which are then read by the mappers, then how can data locality be achieved.
If I understand your question, the method of processing doesn't matter. The JobTracker places tasks based on input locality. So if you are providing the names of the file you want as input as -input, then the JT will use the locations of those blocks. (Remember: streaming.jar is basically a big wrapper around the Java methods and the parameters you pass to it are essentially the same as you'd provide to a "real" Java app.)
Or are you saying your -input is a list of other files to read? In the case, there is no locality. But again, streaming or otherwise makes no real difference.
> Likewise, is there any easier way to make those files accessible other than using the -cacheFile flag?
> That requires building a very very long hadoop command (100s of files potentially). I'm worried about overstepping some command-line length limit...plus it would be nice to do this programatically, say with the DistributedCache.addCacheFile() command, but that requires writing your own driver, which I don't see how to do with streaming.
>
> Thoughts?
I think you need to give a more concrete example of what you are doing. -cache is used for sending files with your job and should have no bearing on what your input is to your job. Something tells me that you've cooked something up that is overly complex. :D
Re: Streaming data locality
Posted by Keith Wiley <kw...@keithwiley.com>.
On Feb 3, 2011, at 9:46 AM, Harsh J wrote:
> Hello,
>
> On Thu, Feb 3, 2011 at 10:46 PM, Keith Wiley <kw...@keithwiley.com> wrote:
>> I've seen this asked before, but haven't seen a response yet.
>>
>> If the input to a streaming job is not actual data splits but simple HDFS file names which are then read by the mappers, then how can data locality be achieved.
>
> Also, if you're only looking to not split the files, you can pass in a
The files won't be split, they're only 6MBs. I'm looking to get the files to my streaming job somehow and the method I've chosen is to send mere fileNAMES via the streaming API and have the streaming program open the file from HDFS through a symbolic link in the distribute cache (the link originating from -cacheFile presumably).
> custom FileInputFormat with isSplitable returning false? You'll lose
> completeness in locality cause of blocks not present in the chosen
> node though, yes -- But I believe that adding a hundred files to
> DistributedCache is not the solution, as the DistributedCache data is
> set to ALL the nodes AFAIK.
My understanding is that the -cacheFile option and the DistributedCache.addCacheFile() method don't copy the entire file to the distributed cache, but rather make tiny symbolic links to the actual HDFS file. Correct? If you don't think I should add 100s of files to the distributed cache (or even 100s of links), then how else can I make the files available to my streaming program?
Put another way, do you know of another method by which to permit the streaming programs to read files from HDFS?
Thanks.
________________________________________________________________________________
Keith Wiley kwiley@keithwiley.com keithwiley.com music.keithwiley.com
"And what if we picked the wrong religion? Every week, we're just making God
madder and madder!"
-- Homer Simpson
________________________________________________________________________________
Re: Streaming data locality
Posted by Harsh J <qw...@gmail.com>.
Hello,
On Thu, Feb 3, 2011 at 10:46 PM, Keith Wiley <kw...@keithwiley.com> wrote:
> I've seen this asked before, but haven't seen a response yet.
>
> If the input to a streaming job is not actual data splits but simple HDFS file names which are then read by the mappers, then how can data locality be achieved.
Also, if you're only looking to not split the files, you can pass in a
custom FileInputFormat with isSplitable returning false? You'll lose
completeness in locality cause of blocks not present in the chosen
node though, yes -- But I believe that adding a hundred files to
DistributedCache is not the solution, as the DistributedCache data is
set to ALL the nodes AFAIK.
--
Harsh J
www.harshj.com
Re: Streaming data locality
Posted by Keith Wiley <kw...@keithwiley.com>.
On Feb 3, 2011, at 9:29 AM, David Rosenstrauch wrote:
> On 02/03/2011 12:16 PM, Keith Wiley wrote:
>> I've seen this asked before, but haven't seen a response yet.
>>
>> If the input to a streaming job is not actual data splits but simple
>> HDFS file names which are then read by the mappers, then how can data
>> locality be achieved.
>>
>> Likewise, is there any easier way to make those files accessible
>> other than using the -cacheFile flag? That requires building a very
>> very long hadoop command (100s of files potentially). I'm worried
>> about overstepping some command-line length limit...plus it would be
>> nice to do this programatically, say with the
>> DistributedCache.addCacheFile() command, but that requires writing
>> your own driver, which I don't see how to do with streaming.
>>
>> Thoughts?
>
> Submit the job in a Java app instead of via streaming? Have a big loop where you repeatedly call job.addInputPath. (Or, if you're going to have a large number of input files, use CombineFileInputFormat for efficiency.)
Well, I know how to write a typical Hadoop driver which "extends Configured implements Tool" if that's what you mean, but then how to I kick off a streaming job from that driver? I only know how to start a "normal" Java Hadoop job that way (via JobClient.runJob(conf);). How do I start a streaming job using that method? I only know how to start a streaming job by launching the streaming jar from the command line?
Does my question make sense?
________________________________________________________________________________
Keith Wiley kwiley@keithwiley.com keithwiley.com music.keithwiley.com
"I do not feel obliged to believe that the same God who has endowed us with
sense, reason, and intellect has intended us to forgo their use."
-- Galileo Galilei
________________________________________________________________________________
Re: Streaming data locality
Posted by David Rosenstrauch <da...@darose.net>.
On 02/03/2011 12:16 PM, Keith Wiley wrote:
> I've seen this asked before, but haven't seen a response yet.
>
> If the input to a streaming job is not actual data splits but simple
> HDFS file names which are then read by the mappers, then how can data
> locality be achieved.
>
> Likewise, is there any easier way to make those files accessible
> other than using the -cacheFile flag? That requires building a very
> very long hadoop command (100s of files potentially). I'm worried
> about overstepping some command-line length limit...plus it would be
> nice to do this programatically, say with the
> DistributedCache.addCacheFile() command, but that requires writing
> your own driver, which I don't see how to do with streaming.
>
> Thoughts?
Submit the job in a Java app instead of via streaming? Have a big loop
where you repeatedly call job.addInputPath. (Or, if you're going to
have a large number of input files, use CombineFileInputFormat for
efficiency.)
HTH,
DR