forked from D-Net/dnet-hadoop
[Actionmanager] remove invalid records from the input graph before groupGraphTableByIdAndMerge
This commit is contained in:
parent
e5ebb500cf
commit
1e423fdc07
|
@ -111,7 +111,9 @@ public class PromoteActionPayloadFunctions {
|
|||
SerializableSupplier<Function<G, Boolean>> isNotZeroFn,
|
||||
Class<G> rowClazz) {
|
||||
TypedColumn<G, G> aggregator = new TableAggregator<>(zeroFn, mergeAndGetFn, isNotZeroFn, rowClazz).toColumn();
|
||||
|
||||
return rowDS
|
||||
.filter((FilterFunction<G>) o -> isNotZeroFn.get().apply(o))
|
||||
.groupByKey((MapFunction<G, String>) x -> rowIdFn.get().apply(x), Encoders.STRING())
|
||||
.agg(aggregator)
|
||||
.map((MapFunction<Tuple2<String, G>, G>) Tuple2::_2, Encoders.kryo(rowClazz));
|
||||
|
|
Loading…
Reference in New Issue