Data handling in map_reduce api

[nadeshseen]

Hello, it would be really helpful if someone could clarify these queries or point me in the right direction, thank you.
Setup - Openwhisk(Compute Backend) + Minio(Storage Backend)

1. Is there a way to get the data in a specific chunk size without partitioning the data before?
   What I can see from the given examples is that all the data for a particular key is read in mem. But this would cause an error. Example - Say, Input data size is of 10GB and my system can support only 10 mappers at a time. So, I want each of them to process 1 GB data. Also, each mapper are using container runtime with 256MB RAM.
2. Is there an API call to get stats of map functions like there is for call_async? I want to know the mem usage of each map function. I gave same input to 2 map functions. One of them could run, but other gave MemoryError: ('HANDLER', 'Function exceeded maximum memory and was killed')".
3. Is there an map reduce function example to store the output of mapper on Storage Backend and work with keys? I want mapper to store data in Minio & reducer to get data from Minio. Because we can't pass input/output more than 5MB directly, right?
4. Is there no partition by key functionality in lithops? This will be useful when output of mappers can't be handled by single reducer, right? Right now even if we use obj_reduce_by_key it won't be useful because it just works on separate input splits and gives separate results for each input split.

[JosepSampe]

Hi @nadeshseen

1. If your data is a txt or csv file (or any file format that has lines that end with \n), you can use the obj_chink_size parameter in the map_reduce() call. See this example
2. the call_async() call returns 1 future, the map() call returns a list of futures, so you only need to iterate the list of futures and get the stats:

...
futures = fexec.map(....)
for f in futures:
    print(f.stats)

3. This is how the map_reduce() call works by default, You don't need to do anything, the data that you return in the return statement of your map function is automatically passed to the reduce function through the storage backend (Minio in your case).

4. If you are analyzing more than 1 object, like in this example, you can use the obj_reduce_by_key parameter in the map_reduce() call to spawn 1 reducer for each object.

In the api definition you can see all the avaiable parameters of the map_reduce() call.
In the examples folder you can find multiple map_reduce examples.

[JosepSampe]

1. Instead of providing obj_chunk_size, you can read the file iterating the data_stream, and providing the amount of bytes you want to read on each iteration, for example: data = obj.data_stream.read(64*1024)
2. In the stats you have func_result_size
3. Check this

[nadeshseen]

1. is there a documentation for this? Also, is there a way to download the whole file and read it line by line? Couldn't find an example here.
2. In map_reduce() api's case func_result_size will return the size of key. Is there a way to get the actual size of the objects stored?
   thanks

[JosepSampe]

1. If you want to download the whole file, you only need to run data = obj.data_stream.read() and then iterate the data.
2. The func_result_size is the size = pickle.dumps(output) of what you return in the return statement of your map function

[nadeshseen]

I have confirmed it by running the following. The data remains in-memory unless I use del(data)

print(int(subprocess.check_output("cat /sys/fs/cgroup/memory/memory.usage_in_bytes", shell=True).decode())/(1024 ** 2))
data = obj.data_stream.read()
print(int(subprocess.check_output("cat /sys/fs/cgroup/memory/memory.usage_in_bytes", shell=True).decode())/(1024 ** 2))

Output is approx 70MB and 170MB. The file I am reading is approx 100MB input.txt file stored in minio.

PS - I want to read the file line by line. So that memory usage remains minimal.

[nadeshseen]

ok got it. I was looking for this download_file() API
Also, is there a way to specify connection timeout value?
If i try to download multiple 1GB files from minio. I get the following error - botocore.exceptions.ConnectTimeoutError: Connect timeout on endpoint URL:
PS - It works fine if I download multiple files of around 100MB.