Spark enrichment integration issues

[Alex-Ikanow]

• inMapPartitions(Post|PrePost)Group fails if any of the partitions receive an empty iterator (fixed by this commit)
• For some reason, the scala inMapPartitions(Post|PrePost)Group methods take java iterables as there input, meaning that it is necessary to insert an extra map stage after the groupBy call, looking something like .map { case (o, it) => (o. it.asScala)
  • (seems like it would be an easy fix to perform that map inside the inMapPartitons* code)
• (UPDATE: SOLVED BUT NOT FIXED - SEE SECOND COMMENT BELOW) According to Yuri, the group key is not inserted into the onObjectBatch calls (looking at the unit tests, that is not tested)
  • Actually in latest reports, .inMapPartitionsPreGroup("key") generates empty keys (in the first element of the return 2-tuple)
    • There is actually test code showing that work though
• Also reported by Yuri, and contradicting test code:

If I use groupBy ( get field) instead of groupByKey, it changes the type from 

Iterator[(com.ikanow.aleph2.data_model.interfaces.data_analytics.IBatchRecord,  Iterable[(Long, com.ikanow.aleph2.data_model.interfaces.data_analytics.IBatchRecord)])], that .inMapPartitionsPostGroup expects to
Iterator[(com.ikanow.aleph2.core.shared.utils.BatchRecordUtils.JsonBatchRecord, Iterable[(com.ikanow.aleph2.data_model.interfaces.data_analytics.IBatchRecord, (Long, com.ikanow.aleph2.data_model.interfaces.data_analytics.IBatchRecord))])] 

It makes sense after I found an obscure reference in the doc: groupBy( func key) treats the source RDD as NOT-Paired. So it pairs it by adding that key. Which of course breaks .inMapPartitionsPostGroup 

See the linked test code, in JavaRDDs I have a groupBy immediately followed by inMapPartitionsPrePostGroup which has the same "input" type as inMapPartitionsPostGroup (code)

[Alex-Ikanow]

(ignore commit ref ending 95, that got pointed to the wrong issue)

[Alex-Ikanow]

OK the mystery of the missing key from .inMapPartitionsPreGroup(key...) is solved

The list of keys gets passed into the next_grouping_keys in onStageInitialize, but unless the module happens to use those fields to generate a key and pass it as the 3rd arg of the emitMutableObject request, it is ignored

You could put some code here to override the case where next_grouping_fields is set (needs to be saved) but the output key is Optional.empty

Note that the Hadoop code (which uses the grouping_fields field of the control param) is here for reference.

• (ah I see - this is just used to generate the reduce key in Hadoop, but isn't passed anywhere else, the override value is used first, then the grouping fields as a "fallback")

As a workaround in the meantime you just have to include the key by hand in the emit