[
https://issues.apache.org/jira/browse/FLINK-2237?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15147212#comment-15147212
]
ASF GitHub Bot commented on FLINK-2237:
---------------------------------------
Github user fhueske commented on the pull request:
https://github.com/apache/flink/pull/1517#issuecomment-184167321
Hi Gabor,
thanks for the updates and additions you made to the PR. If didn't go
through them yet, but would like to follow up on the discussion and propose a
few potential improvements.
I agree that using multiple partitions would make the implementation more
complex and the reasoning about the behavior less straightforward. The
motivation for partitions is twofold. First, it should helps to maintain a
continuous data flow without backpressure effects. Second, it should improve
the effectiveness of the combiner by avoiding to evict hot elements which are
frequently updated which would happen for each full eviction of the table.
Emitting the partition with the least segments wouldn't be a good strategy for
the reasons you mentioned. However, a strategy could be to remove the partition
with the smallest updates / keys ratio. However, this would require more
thoughts…
Independent of the partitions discussion, I have a few ideas to make the
compaction more effective. However, these would break the compatibility with
the AbstractMutableHashTable interface. Maybe we can provide a common base
class and special implementations for the SolutionSet case and the Combiner
case?
1. Remember a pointer to the hole which is closest to the start of the
RecordArea (or partition).
2. When compacting, emit all elements which are before the first hole.
These elements are "cold" and have not been updated since the last compaction
(assuming non-in-place updates, but in-place updates should not be compacted
anyways). Compact all elements after the first hole as usual. The benefit of
this is to free more segments by compaction which delays (or even completely
avoids) full eviction, and thus improving the compaction rate for hot elements.
3. Do only compact if more than (for instance) 50% of space can be freed
and do a full eviction otherwise. If there are too few holes in the table, most
elements have not been touched (in case of non-in-place updates). Hence, we
should just emit everything.
We can also think about adding a by-pass switch if the HT observes that the
combiner does not help to significantly reduce the data (too many unique keys
for not enough memory) but only adds additional overhead. For instance if the
number of records is not reduced by 2x, we could simply forward the records to
the collector (In case of partitions, this decision could be done on partition
level).
> Add hash-based Aggregation
> --------------------------
>
> Key: FLINK-2237
> URL: https://issues.apache.org/jira/browse/FLINK-2237
> Project: Flink
> Issue Type: New Feature
> Reporter: Rafiullah Momand
> Assignee: Gabor Gevay
> Priority: Minor
>
> Aggregation functions at the moment are implemented in a sort-based way.
> How can we implement hash based Aggregation for Flink?
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
