It would be useful to have some way to open multiple files at once into a
single RDD (e.g. sc.textFile(iterable_over_uris)). Logically, it would be
equivalent to opening a single file which is made by concatenating the
various files together. This would only be useful, of course, if the source
files were all in the same format.
Nick
On Mon, Apr 28, 2014 at 7:40 PM, Andrew Ash <and...@andrewash.com> wrote:
> The way you've written it there, I would expect that to be serial runs.
> The reason is, you're looping over matches with a driver-level map, which
> is serialized. Then calling foreachWith on the RDDs executes the action in
> a blocking way, so you don't get a result back until the cluster finishes.
>
> You can have multiple jobs running at the same time though by sharing a
> SparkContext among threads. Rather than run all the foreachWith()s in
> serial on a single thread in the driver, try running each in its own thread.
>
>
>
>
> On Tue, Apr 29, 2014 at 1:35 AM, Art Peel <found...@gmail.com> wrote:
>
>>
>> I’m trying to process 3 S3 files in parallel, but they always get
>> processed serially.
>> Am I going about this the wrong way? Full details below.
>>
>> Regards,
>> Art
>>
>>
>>
>> I’m trying to do the following on a Spark cluster with 3 slave nodes.
>>
>> Given a list of N URLS for files in S3 (with s3n:// urls),
>>
>> For each file:
>> 1. search for some items
>> 2. record the findings from step 1 in an external data store.
>>
>> I’d like to process the 3 URLs in parallel on 3 different slave nodes,
>> but instead, they are getting processed serially on one node.
>>
>> I’ve tried various versions of my code to no avail. I’m also creating the
>> SparkContext with "spark.scheduler.mode", "FAIR”.
>>
>> Have I made a fundamental mistake?
>>
>> I’m running Spark 0.9.1 and my code looks roughly like this:
>>
>> def processFiles(sc: SparkContext) {
>>
>> val urls = List("s3n://foo/file.txt", "s3n://bar/file.txt",
>> "s3n://baz/file.txt")
>>
>> val hadoopFiles = urls.map(url => {
>> sc.hadoopFile(url, classOf[FooInputFormat], classOf[LongWritable],
>> classOf[WritableFooRecord])
>> })
>>
>> val matches = hadoopFiles.par.map((hadoopFile) => {
>>
>> findMatches(hadoopFile)
>> })
>>
>>
>> matches.map((matchRDD) => {
>> recordMatches(matchRDD)
>> })
>> }
>>
>>
>> def findMatches(hadoopFile: RDD): RDD = {
>>
>> hadoopFile.map( record => caseClassResultFromSearch(record) )
>>
>> }
>>
>> def recordMatches(matchRDD: RDD) {
>>
>>
>> matchRDD.foreachWith(_ => {
>>
>> makeRestClient(config)
>>
>> // I get 3 jobs referring to the line number of the next line, but
>> the jobs run serially on one node.
>> })((matchRecord, client) => {
>>
>> client.storeResult(matchRecord)
>>
>> }
>>
>>
>> }
>>
>>
>>
>
