[WIP] pyDKB: stage fix

[mgolosova]

There were some issues with Stage:

• when running with empty cmdline arguments list;
• when interrupting stage, taking filenames from STDIN;
• when runnig with custom EOM, only first message was read.

The PR fixes these issues.

To check the behaviour I suggest to use test stage in Utils/Dataflow/test/pyDKB.
With which set of parameters to run this stage and what to expect [3]:

1. Local files:

• without parameters (should claim that no input files specified);
• -m f (should claim that no input files specified);
• -m f input/NDjson.json (should produce output file input/out/NDjson.ttl [1] or claim that file exists);
• input/NDjson.json (should produce output file input/out/NDjson.ttl or claim that file exists);
  I believe that first two items shows that -m f and "without parameters" are the same, so 3rd and 4th are the same too.
• -i ./input NDjson.json (should produce output file input/out/NDjson.ttl or claim that file exists);
• -i ./input (should produce output file input/out/NDjson.ttl (or claim that file exists) and claim that fail reading messages from jsonArray.json);
• -o ./output input/NDjson.json (should produce output file output/NDjson.ttl or claim that the file exists);
• -o output input/NDjson.json (should produce output file input/output/NDjson.ttl or claim that the file exists);
• -d s input/NDjson.json (should write output to STDOUT);
• -s s (should take data from STDIN and write output to out/???.ttl files [2]);

2. HDFS files:

• --hdfs (should claim that no input files specified);
• --hdfs /path/to/hdfs/file.json (should produce output file /path/to/hdfs/out/file.ttl in HDFS or claim that the file already exists);
• -m f --hdfs /path/to/hdfs/file.json (should produce output file /path/to/hdfs/out/file.ttl in HDFS or claim that the file already exists);
• -d h input/NDjson.json (should produce output file /user/DKB/temp/out/NDjson.ttl in HDFS or claim that the file already exists);
• -s h /path/to/hdfs/file.json (should produce output file out/file.ttl or claim that the file already exists);
• --hdfs -o /path/to/hdfs/dir /path/to/hdfs/file.json (should produce output file /path/to/hdfs/dir/file.ttl in HDFS or claim that the file already exists);
• -s h -i /path/to/hdfs file.json (should process hdfs:///path/to/hdfs/file.json and produce local output file out/file.ttl or claim that the file already exists);
• -s h -i /path/to/hdfs (should produce local out/*.ttloutput files in current directory for each *.json file from the HDFS directory or claim that the file already exists);

3. Streaming mode:

• for every run in streaming mode (-m s) should add EOP marker to STDOUT;
• -m s (should run in stream mode: take input data from STDIN and write output data to STDOUT);
• -m s -d f (should take input data from STDIN and write output data to ./out/???.ttl ( == -s s == -m f -s s));
• -m s -d h (should take input data from STDIN and write output data to /user/DKB/temp/out/???.ttl ( == -s s -d h == --hdfs -s s));
• -m s -s f (should claim that no input files is specified ( == -s f -d s == -d s == -m f -d s));
• -m s -s h (should claim that no input files is specified ( == -s h -d s == --hdfs -d s == -m f -s h -d s));

4. MapReduce mode:

• -m m (should run in MapReduce mode: take input data from STDIN (line by line) and write output data to STDOUT ( ~= -m s, except that no EOP marker appears in output);
• -m m -s s (== -m m);
• -m m -d s (== -m m);
• -m m -s h (should take input HDFS file names from STDIN and write output data to STDOUT);
• -m m -d h (should take input from STDIN and write output data to /user/DKB/temp/out/???.ttl);
• -m m --hdfs (should take input HDFS file names from STDIN and write output data to /user/DKB/temp/out/???.ttl);
• -m m -s f (should claim that given source is not supported in (m)ap-reduce mode);
• -m m -d f (should claim that given destination is not supported in (m)ap-reduce mode);

5. Custom EOP/EOM:

• -d s input/EOMDjson.json (should fail to read input data);
• -d s -e EOM input/NDjson.json (should fail to read input data);
• -d s -e EOM input/EOMDjson.json (should output processed data to STDOUT with same EOM);
• -d s -E EOP input/NDjson.json (should output EOP after every message);
• -d s -e '' input/jsonArray.json (should output processed data to STDOUT, also in form of JSON array (or hash, if there`s only one message)) [4];
• -d s -e '' input/NDjson.json (should fail to read input data);

Not sure if this list is complete and that "expected" behaviour is intuitive, so if looking on cmdline parameters one expects different behaviour, please tell.

[1] NDjson.ttl must contain (Pythonic) string representation of JSON strings from NDjson.json.
[2] ??? is timestamp (seconds) of the moment when the filename is being chosen.
[3] Stage help message was changed to eliminate disputable moments; so everything said above is to be reviewd according to it.
[4] Seems like there`s a problem: if EOM == '', it allows to read JSON array -- but how output messages should be delimited? Maybe they should be collected into a single JSON array, too? (solved in PR #171)

---

ToDo:

• rewrite checklist according to the new edition of the stage help message;
• move help message improvements into a separate PR;
• add input data sample with EOM other than '\n';
• add test utilities:
  • to check ARGS parameter values after *Stage.args_parse();
  • to check expected behaviour of the stage as a whole;
• fix everything not passing tests;
• review changes;
• HDFS mode:
  • add tests;
  • fix everything that works not the way it is expected (or adjust expectations ;) );
  • review changes.

---

Waits for: #149 (merged)
Stage help/usage message changes are copied to a separate PR (#171) (merged)
Some of the errors are going to be fixed in #172 (re-refactoring).

[Evildoor]

It seems that there is a typo in get_readable_stream description: "nearset" instead of "nearest".

[mgolosova]

@Evildoor,
thank you, now fixed.

Found out that after all the changes test stage still does not work as expected. Details are in the comment below.

[mgolosova]

Checked the test stage before merging and here is what I saw:

1. Running without parameters produces error:

git.dkb-project/Utils/Dataflow/test/pyDKB$ ./json2TTL.py 
(WARN) pyDKB.dataflow.cds failed (No module named invenio_client.contrib)
Traceback (most recent call last):
  File "./json2TTL.py", line 75, in <module>
    main(sys.argv[1:])
  File "./json2TTL.py", line 56, in main
    stage.configure(args)
  File "./../../pyDKB/dataflow/stage/ProcessorStage.py", line 181, in configure
    .setType(self.__input_message_type) \
  File "./../../pyDKB/dataflow/communication/consumer/__init__.py", line 62, in build
    instance = self.consumerClass(config)
  File "./../../pyDKB/dataflow/communication/consumer/Consumer.py", line 33, in __init__
    self.reconfigure()
  File "./../../pyDKB/dataflow/communication/consumer/FileConsumer.py", line 35, in reconfigure
    self.init_sources()
  File "./../../pyDKB/dataflow/communication/consumer/FileConsumer.py", line 45, in init_sources
    if not self.input_files:
AttributeError: 'FileConsumer' object has no attribute 'input_files'

2. MapReduce mode should take file names from STDIN, but it does not work:

git.dkb-project/Utils/Dataflow/test/pyDKB$ ./json2TTL.py -m m
(WARN) pyDKB.dataflow.cds failed (No module named invenio_client.contrib)
input/NDjson.json
(WARN) (InputStream) Failed to read input message as JSON.
(==) Cause: No JSON object could be decoded
(==) Original message: 'input/NDjson.json
(==) '

I think I`m gonna add a checklist to the PR description with all the usecases to be tested before asking for a review, approve or merge. Without it we risk to miss something again.

[mgolosova]

I`m closing this PR, as most of the fixes are done in the series of pyDKB-fix PRs (see #197).
HDFS-related functionality is still not reviewed, but to be done later (see https://trello.com/c/93c1wsce).

This PR could be left open as a reminder, but I don`t think it is a good idea: we have other tools for task tracking, after all.