Replace buggy pig rank function with custom solution

[marekhorst]

This issue was originally reported in https://github.com/openaire/iis/issues/927 but since it requires changes in CoAnSys PIG script I am reporting it once again here.

Pig RANK operation related problems ware mitigated within OpenAIRE scope several times already: either by extending the amount of memory (https://github.com/openaire/iis/issues/796, https://github.com/openaire/iis/issues/807) or by refactoring PIG script to minimize memory footprint during the RANK operation (https://github.com/CeON/CoAnSys/issues/425).

After recent increase in number of publications (to 37M) we are struggling again with the memory related problem:

java.lang.OutOfMemoryError
    at java.io.ByteArrayOutputStream.hugeCapacity(ByteArrayOutputStream.java:123)
    at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:117)
    at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
    at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:153)
    at java.io.DataOutputStream.write(DataOutputStream.java:107)
    at java.io.DataOutputStream.writeUTF(DataOutputStream.java:401)
    at java.io.DataOutputStream.writeUTF(DataOutputStream.java:323)

full log is available here: https://pastebin.com/dk2C8wLF

Pig execution plan is available here: https://pastebin.com/bAUsCNjb

claiming again RANK operation to be the phase when the map task failed:

Failed Jobs:
JobId   Alias   Feature Message Outputs
job_1524597382992_21544 wc_ranked   ORDER_BY    Message: Job failed!   

[marekhorst]

Apparently there is an alternative processing path within COANSYS documents similarity involving custom ranking. The problem is this script at some point became incompatible with the rest of the documents similarity algorithm (as described in https://github.com/CeON/CoAnSys/issues/427) but we could definitely use custom rank serializer and replace the following line causing problems:

wc_ranked = rank wc by count asc;

with custom ranking solution:

wc_tmp = order wc by count asc parallel 1;
STORE wc_tmp INTO '$outputPath$WORD_RANK_HR' using pl.edu.icm.coansys.similarity.pig.serializers.RankStorage();
wc_ranked = LOAD '$outputPath$WORD_RANK_HR' as (rank_num:long, count:long, term:chararray);

Firsts tests on 37M (and on 103M) of documents proved the proposed solution eliminates the memory related issue.