How to call a defined function with map()?

[byr1288]

Hello,

I am trying to call a defined function with the map(). I understand that that the map function appl the same procedure to every item in an iterable data structure. But the pre-defined is not being called.

I have a defined function, like this:

def process (records):
   fields = records.map(lambda x: x.split('ID:')[0].split('\n'))
   return fields;
 

Here is the map() and the defined function being called:

files = sc.wholeTextFiles('file:///data/*/*')
records = files.map(lambda x: x[1])
results = records.map(lambda x: process(x))
 

There are 4,560 files loaded up by wholeTextFiles. Perhaps, I am missing something basic. Can you help?

[dwblas]

List comprehension may work better

fields=[x.split('ID:')[0].split('\n') for x in records] 

but since you don't open and read the files (don't know what wholeTextFiles does) you won't have any data for list comprehension/map.

[byr1288]

Hi dwblas, thanks for your reply. WholeTextFiles = Read a directory of text files from HDFS, a local file system (available on all nodes), or any Hadoop-supported file system URI. https://spark.apache.org/docs/2.0.2/...rkContext.html

It returns the file name and the contents of the file, a pairRDD.

I did a count on “records” and there are 4,650 files with data, so there is something there in “records”.

I am new to Python. If I had a RDD, how would I call a predefined function with the map(), so I can iterate through each data set in the RDD? This does not work: results = records.map(lam bda x: process(x))

[dwblas]

What happened when you tried the code I posted. This list comprehension replaces

records = files.map(lambda x: x[1])

so hopefully it will now work but I am shooting in he dark as don't know what files and records contains.

records=[x[1] for x in files] 

[byr1288]

Hi Dwblas, I was able to figure this out. I was calling a defined function that has a map() with the map(). I experienced that this call errors. I rewrote my defined function that does not include the map().

files = sc.wholeTextFiles ('file:///data/*/*')
records = files.map(lambda x: x[1])
results = records.map(lambda x: process_records(x))