-
Notifications
You must be signed in to change notification settings - Fork 95
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Performance optimization for aggregates: do not save context of varia…
…bles in state
- Loading branch information
Showing
12 changed files
with
145 additions
and
185 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
37 changes: 0 additions & 37 deletions
37
.../nussknacker/engine/flink/util/transformer/aggregate/AddedElementContextStateHolder.scala
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
23 changes: 0 additions & 23 deletions
23
...a/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/OutputContextStrategy.scala
This file was deleted.
Oops, something went wrong.
46 changes: 16 additions & 30 deletions
46
...ouk/nussknacker/engine/flink/util/transformer/aggregate/UnwrappingAggregateFunction.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,54 +1,40 @@ | ||
| package pl.touk.nussknacker.engine.flink.util.transformer.aggregate | ||
|
|
||
| import org.apache.flink.api.common.functions.AggregateFunction | ||
| import org.apache.flink.api.java.tuple | ||
| import org.apache.flink.api.java.tuple.Tuple2 | ||
| import pl.touk.nussknacker.engine.api.typed.typing.TypingResult | ||
| import pl.touk.nussknacker.engine.api.{Context, ValueWithContext} | ||
| import pl.touk.nussknacker.engine.flink.util.transformer.aggregate.UnwrappingAggregateFunction.AccumulatorWithContext | ||
| import pl.touk.nussknacker.engine.flink.util.keyed.{KeyEnricher, StringKeyedValue} | ||
|
|
||
| /** | ||
| * This class unwraps value from input's KeyedValue. It also accumulate first Nussknacker's context that will be passed in output at the end. | ||
| * This class unwraps value from input's KeyedValue. It also accumulate key that will be passed in output at the end. | ||
| * | ||
| * NOTE: it would be much cleaner if we evaluated aggregateBy here. However, FLINK-10250 prevents us from doing this and we *have* to compute it beforehand | ||
| * | ||
| * When using this class it's important that agggregator, passedType and unwrap must match: unwrap result is of passedType and can be processed by aggregator | ||
| */ | ||
| object UnwrappingAggregateFunction { | ||
| //We use Tuple2 here, to create TypeInformation more easily | ||
| type AccumulatorWithContext = Tuple2[AnyRef, Context] | ||
| } | ||
|
|
||
| class UnwrappingAggregateFunction[T](aggregator: Aggregator, | ||
| passedType: TypingResult, | ||
| unwrap: T => AnyRef, | ||
| outputContextStrategy: OutputContextStrategy) | ||
| extends AggregateFunction[ValueWithContext[T], Tuple2[AnyRef, Context], ValueWithContext[AnyRef]] { | ||
| class UnwrappingAggregateFunction[Input](aggregator: Aggregator, | ||
| passedType: TypingResult, | ||
| unwrapAggregatedValue: Input => AnyRef) | ||
| extends AggregateFunction[ValueWithContext[StringKeyedValue[Input]], StringKeyedValue[AnyRef], ValueWithContext[AnyRef]] with KeyEnricher { | ||
|
|
||
| private val expectedType = aggregator.computeOutputType(passedType) | ||
| .valueOr(msg => throw new IllegalArgumentException(msg)) | ||
|
|
||
| override def createAccumulator(): AccumulatorWithContext = new Tuple2(aggregator.createAccumulator(), null) | ||
| override def createAccumulator(): StringKeyedValue[AnyRef] = StringKeyedValue(null, aggregator.createAccumulator()) | ||
|
|
||
| override def add(value: ValueWithContext[T], accumulator: AccumulatorWithContext): AccumulatorWithContext = { | ||
| val underlyingAcc = aggregator.add(unwrap(value.value), accumulator.f0) | ||
| val contextToUse = outputContextStrategy.transform(Option(accumulator.f1), value.context) | ||
| new Tuple2(underlyingAcc, contextToUse.orNull) | ||
| override def add(wrappedInput: ValueWithContext[StringKeyedValue[Input]], accumulator: StringKeyedValue[AnyRef]): StringKeyedValue[AnyRef] = { | ||
| wrappedInput.value.mapValue(input => aggregator.add(unwrapAggregatedValue(input), accumulator.value)) | ||
| } | ||
|
|
||
| override def getResult(accumulator: AccumulatorWithContext): ValueWithContext[AnyRef] = { | ||
| val accCtx = Option(accumulator.f1).getOrElse(outputContextStrategy.empty) | ||
| val finalResult = aggregator.alignToExpectedType(aggregator.getResult(accumulator.f0), expectedType) | ||
| ValueWithContext(finalResult, accCtx) | ||
| override def getResult(accumulator: StringKeyedValue[AnyRef]): ValueWithContext[AnyRef] = { | ||
| val finalResult = aggregator.alignToExpectedType(aggregator.getResult(accumulator.value), expectedType) | ||
| ValueWithContext(finalResult, enrichWithKey(Context(""), accumulator)) | ||
| } | ||
|
|
||
| override def merge(a: AccumulatorWithContext, b: AccumulatorWithContext): AccumulatorWithContext = { | ||
| val underlyingAcc = aggregator.merge(a.f0, b.f0) | ||
| val firstContext = Option(a.f1).getOrElse(b.f1) | ||
| new tuple.Tuple2[AnyRef, Context](underlyingAcc, firstContext) | ||
| override def merge(a: StringKeyedValue[AnyRef], b: StringKeyedValue[AnyRef]): StringKeyedValue[AnyRef] = { | ||
| val mergedKey = Option(a.key).getOrElse(b.key) | ||
| val mergedValue = aggregator.merge(a.value, b.value) | ||
| StringKeyedValue(mergedKey, mergedValue) | ||
| } | ||
|
|
||
| } | ||
|
|
||
|
|
||
|
|
Oops, something went wrong.