renku run error - too many dependent files

[naspert]

Trying to pre-process an audio dataset with 30000 files and merge them into a single hdf5 makes the following error:

Renku version: 0.5.0
OS: Linux (#1 SMP Tue Nov 27 12:40:57 EST 2018)
Python: 3.6.8

Traceback

Traceback (most recent call last):
  File "[...]/renku/cli/_exc.py", line 100, in main
    result = super().main(*args, **kwargs)
  File "[...]/click/core.py", line 717, in main
    rv = self.invoke(ctx)
  File "[...]/click/core.py", line 1137, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "[...]/click/core.py", line 956, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "[...]/click/core.py", line 555, in invoke
    return callback(*args, **kwargs)
  File "[...]/renku/cli/_client.py", line 80, in new_func
    result = ctx.invoke(method, client, *args, **kwargs)
  File "[...]/click/core.py", line 555, in invoke
    return callback(*args, **kwargs)
  File "[...]/renku/cli/run.py", line 187, in run
    for _, path in tool.iter_input_files(client.workflow_path)
  File "[...]/renku/api/storage.py", line 122, in pull_paths_from_storage
    stderr=STDOUT,
  File "[...]/subprocess.py", line 423, in run
    with Popen(*popenargs, **kwargs) as process:
  File "[...]/subprocess.py", line 729, in __init__
    restore_signals, start_new_session)
  File "[...]/subprocess.py", line 1364, in _execute_child
    raise child_exception_type(errno_num, err_msg, err_filename)
OSError: [Errno 7] Argument list too long: 'git'

[rokroskar]

Can you specify directories instead of files?

[naspert]

I tried to figure out how to try your suggestion but failed so far. The dataset is the one located at https://github.com/soerenab/AudioMNIST
I then created a script that will read every file from it compute mfcc features and store the result into a single hdf5 output file. (but my script uses a data directory as parameter, not list of files)

[rokroskar]

Hi @naspert thanks for following up - could you share a simple snippet that will reproduce the problem?

[naspert]

# -*- coding: utf-8 -*-

import numpy as np
import glob
import os
import sys
import scipy.io.wavfile as wavf
import scipy.signal
import h5py
import json
import librosa
import multiprocessing
from joblib import Parallel, delayed
import argparse
from tqdm import tqdm

def _wav2mfcc(wav):
    fs = 8000
    return np.pad(librosa.feature.mfcc(wav.T.flatten(), sr=fs), pad_width=((0, 0), (0, 4)), mode='constant')

def wav2mfcc(wav):
    num_cores = multiprocessing.cpu_count()
    s = wav.shape
    res = Parallel(n_jobs=num_cores, verbose=1)(delayed(_wav2mfcc)(wav[k, :]) for k in range(s[0]))
    v = np.stack(res)
    return np.reshape(v, (v.shape[0], v.shape[1], v.shape[2], 1))

def preprocess_data(src, src_meta, n_processes=15):

  
    folders = []
    
    for folder in os.listdir(src):
        # only process folders
        if not os.path.isdir(os.path.join(src, folder)):
            continue
        folders.append(folder)

    pool=multiprocessing.Pool(processes=n_processes)
    res =pool.map(_preprocess_data, [(os.path.join(src, folder), src_meta) for folder in sorted(folders)])
    
    return {k: v for d in res for k, v in d.items()}



def _preprocess_data(src_meta):

   
    src, src_meta= src_meta
    res = {}
    print("processing {}".format(src))

    metaData = json.load(open(src_meta))

    # loop over recordings
    for filepath in sorted(glob.glob(os.path.join(src, "*.wav"))):

        # infer sample info from name
        dig, vp, rep = filepath.rstrip(".wav").split("/")[-1].split("_")
        # read data
        fs, data = wavf.read(filepath)
        # resample
        data = librosa.core.resample(y=data.astype(np.float32), orig_sr=fs, target_sr=8000, res_type="scipy")
        # zero padding
        if len(data) > 8000:
            raise ValueError("data length cannot exceed padding length.")
        elif len(data) < 8000:
            embedded_data = np.zeros(8000)
            offset = np.random.randint(low = 0, high = 8000 - len(data))
            embedded_data[offset:offset+len(data)] = data
        elif len(data) == 8000:
            # nothing to do here
            embedded_data = data
            pass

        ##### AudioNet #####
        
        embedded_data /= (np.percentile(embedded_data, 95) + 0.001)
        tmp_X = np.zeros([1, 1, 1, 8000])
        tmp_X[0, 0, 0] = embedded_data
        label = np.array([[int(dig), 0 if metaData[vp]["gender"] == "male" else 1]])
        key = "{}_{}_{}".format(dig, vp, rep)
        res[key] = {'data':tmp_X, 'label':label}
        
    return res


def merge_data(src, dst, wavs):
     # create folder for hdf5 files
    if not os.path.exists(dst):
        os.makedirs(dst)
   
    number_of_files = 30000
    data_tensor = np.zeros((number_of_files, 8000))
    label_tensor = np.zeros((number_of_files, 3))
    file_index = 0
    
    for folder in os.listdir(src):
        subfolder = os.path.join(src, folder)
        if os.path.isdir(subfolder):
            #print("Processing "+subfolder)
            for file in os.listdir(subfolder):
                if file.endswith('.wav'):
                    filename = os.path.join(subfolder, file)
                    dig, vp, rep = file.rstrip(".wav").split("/")[-1].split("_")
                    key = "{}_{}_{}".format(dig, vp, rep)
                    data = wavs[key]['data']
                    labels = wavs[key]['label']
                    data_tensor[file_index, :] = data
                    label_tensor[file_index, 0:2] = labels # first is digit (0 - 9) and second is gender (0 - male, 1 - female)
                    label_tensor[file_index, 2] = folder
                    file_index += 1
                    if file_index % 1000 == 0:
                        print(str(file_index) + '/30000')
    
    # create mfcc
    data_mfcc = wav2mfcc(data_tensor)
    with h5py.File(os.path.join(dst, 'audiomnist.hdf5'), "w") as f:
        f['data'] = data_tensor
        f['label'] = label_tensor
        f['mfcc'] = data_mfcc
        f['fs'] = 8000
        f['num_classes_label_1_digit'] = 10
        f['num_classes_label_2_gender'] = 2
        f['num_classes_label_3_speaker'] = 60

if __name__ == "__main__":

    parser = argparse.ArgumentParser()
    parser.add_argument('--source', '-src', default=os.path.join(os.getcwd(), "data"), help="Path to folder containing each participant's data directory.")
    parser.add_argument('--destination', '-dst', default=os.path.join(os.getcwd(), "preprocessed_data"), help="Destination where preprocessed data shall be stored.")
    parser.add_argument('--meta', '-m', default=os.path.join(os.getcwd(), "data", "audioMNIST_meta.txt"), help="Path to meta_information json file.")

    args = parser.parse_args()

    # preprocessing
    res = preprocess_data(src=args.source, src_meta=args.meta)
    print("all wavs loaded len={}".format(len(res)))
    merge_data(args.source, args.destination, res)

[naspert]

requirements.txt

[naspert]

(preprocessing script adapted mostly from https://github.com/soerenab/AudioMNIST/blob/master/preprocess_data.py )

[rokroskar]

Hi @naspert apologies for the silence on this issue - Would it be possible to make the hdf5 file in batches? Here it looks like we're running into a git limitation...

[naspert]

Yes that would be the obvious fix on my side (will try soon). On the renku side, would it be possible to split the git call into multiple ones to avoid the overflow ? [Just as a side note, this is a very common use case for us (dataset made of many files)]

[rokroskar]

Would be nice to be able to do that for the user - however, that would separate the commit that creates the output (e.g. your hdf5 file) from the commit that contains a subset of the inputs.

[naspert]

Would it be possible to make the hdf5 file in batches? Here it looks like we're running into a git limitation...

Yes but it becomes cumbersome: you cannot process a single folder at a time (500 files). So it means you have to process very small batches, and then aggregate them together. A fair amount of code for a simple task. Maybe using git for a storing just the metadata about the dependency graph would help here (but that means a different paradigm)

[rokroskar]

I'll need to look into the limitations of git in terms of file numbers some more - but it seems to me that 500 should not be a problem.

[leafty]

Not a git problem.

renku-python/renku/api/storage.py

Line 119 in 450898b

self._CMD_STORAGE_PULL + [shlex.quote(','.join(paths))],

The generated string hits OS limits for argument size when invoking a program: https://unix.stackexchange.com/questions/120642/what-defines-the-maximum-size-for-a-command-single-argument.