create subparser, let's compartmentalize data and ml pipeline -> I'd suggest commands should be extract_features and then train_model (for now only add the first and a TODO for the second, if you have better name pls propose)

not sure about this bit. should I create two methods like add_extract_features(parser) and add_train_model_args(parser) and add the respective options inside of them?

i think for one you can omit the second one, as no logic for it exists. anywaay yeah, you can do something like:

subparsers = parser.add_subparsers(help="Commands", dest="command")
extract_parser = subparsers.add_parser(
            "extract", help="Extract features from input repositories", 
formatter_class=args.ArgumentDefaultsHelpFormatterNoNone)

I do not know how to do this, I've pushed my trial (which does not work) but I'm not sure how these hierarchies of parser work, the rest should be solved.

*keep-freq and initialize the parser like this for the conversion to keep_freq

from sourced.ml.cmd import ArgumentDefaultsHelpFormatterNoNone
...

parser = argparse.ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatterNoNone)

*"Output path for the Code2VecFeatures model "

*Code2VecFeatures , we will proabably have a second Model for the trained ML model

rename r to rows and cache after reduceByKey or you will do it twice, which is time consuming, dont forget to uncache after both collects

not sure how things are cached / uncached in Spark I'm pretty new to it, could you give more details?

sure. in spark you can either do transformations or actions on RDDs/Dataframes/etc. Those objects are lazily evaluated, which means transformations are not applied if an action does not trigger them:

r.map(lambda x: x[0]).filter(my_condition)  # does nothing 
r.map(lambda x: x[0]).filter(my_condition).collect()  # the action collect triggers transformations  

in order to not repeat parts of the pipeline, you can cache the data in RDD or on disk, here:

rows = rows \
    .flatMap(self._get_path) \
    .reduceByKey(operator.add) \
    .persist()  # nothing happens

values = rows.filter(lambda x: type(get_elem(x)) == str).collect() 
# after this, rows is cached, so doing the second collect starts from 
# after the reduceByKey, not before the flatmap
paths = r.filter(lambda x: type(get_elem(x)) == tuple).collect()
rows.unpersist(). # or this will stay in memory/disk

we usually add a persist argument to select level of persistence, for more see here

collect directly here, same for paths

you should not do this here, using the mapping this way means all the computation to map values/context_paths to index are done in the driver, not using Spark. You can do it in different ways, but simplest is to create the indexes after the collect, broadcast them, then use them as mappings before collecting the (doc, [path_context_1, ...]) RDD.

if you want to do this, you need to map the RDD to (doc, [pc]), 1 no ? In all cases I think for now it is best to not add calculations for document/featurefreqs, so simply do a distinct after the map (distinct should be done after mapping of the path-contrxes strings to indexes, see comment above)

the idea was to add the lists for each key to have a (doc, [pc_1, pc_2, ...]) because _doc2pc is a single path context lik (doc, [pc])

okay i see now, problem is this might give you cases like this: (doc, [pc1, pc2, pc1, pc3 ...]) so you ought to still do a distinct, before the reduceByKey

same add features here

*keep-freq

*max-width

*max-length

this should be added to the subparser

take out default value, and use required=True

this is a problem if keep_freq is True, in one case you get the dictionnaries but in the other you get (key, value): index. I would move the index creation to the call, and use self.docfreq to potentially create the frequency dict, and pass it as an optional dict parameter for the Model that default to None.

nest this function in build_doc2pc, and broadcast indexes:

def build_doc2pc(self, rows: RDD):
     value2index = self.sc.broadcast(self.value2index)
     path2index = self.sc.broadcast(self.path2index)
     def _doc2pc(self, row: Row):
            (u, path, v), doc = Vocabulary2Id._unstringify_path_context(row), row[0][1]
             return doc, (value2index.value[u], path2index.value[path], value2index.value[v])
    
    ....
    value2index.unpersist(blocking=True)
    path2index.unpersist(blocking=True)

add sparkContext to __init___ -> self.sc used later

here add root.sc to initialize transformer with sparkContext

I get this error:

  File "/home/hydra/projects/code2vec/src/../src/code2vec.py", line 26, in code2vec
    .link(Vocabulary2Id(root.sc, args.keep_freq, args.output)) \
AttributeError: 'Engine' object has no attribute 'sc'

dont use split_strings on path2index, use it on path_contexts

use merge_strings here for path_contexts

*model.

add:

extract_parser.set_defaults(handler=code2vec_extract_features)

replace this line with:

    try:
        handler = args.handler
    except AttributeError:
        def print_usage(_):
            parser.print_usage()

        handler = print_usage
return handler(args)

you are broadcasting frequencies uselessly here, better to use dict comprehension beforehand:

value2index = self.sc.broadcast({key: id for key, (id, _) in value2index_freq.items()})