I`ve implemented a combiner [1] in Flink by extending OneInputStreamOperator in Flink. I call my operator using "transform". It works well and I guess it is useful if I import this operator in the DataStream.java. I just need more to check if I need to touch other parts of the source code.
But now I want to tackle data skew by altering the way Flink partition keys using KeyedStream. [1] https://felipeogutierrez.blogspot.com/2019/08/implementing-dynamic-combiner-mini.html *--* *-- Felipe Gutierrez* *-- skype: felipe.o.gutierrez* *--* *https://felipeogutierrez.blogspot.com <https://felipeogutierrez.blogspot.com>* On Mon, Sep 23, 2019 at 2:37 PM Biao Liu <mmyy1...@gmail.com> wrote: > Hi Felipe, > > If I understand correctly, you want to solve data skew caused by > imbalanced key? > > There is a common strategy to solve this kind of problem, pre-aggregation. > Like combiner of MapReduce. > But sadly, AFAIK Flink does not support pre-aggregation currently. I'm > afraid you have to implement it by yourself. > > For example, introducing a function caching some data (time or count > based). This function should be before "keyby". And it's on a non-keyed > stream. It does pre-aggregation just like what the aggregation after > "keyby" does. In this way, the skewed keyed data would be reduced a lot. > > I also found a suggestion [1] from Fabian, although it's long time ago. > > Hope it helps. > > 1. > https://stackoverflow.com/questions/47825565/apache-flink-how-can-i-compute-windows-with-local-pre-aggregation > > Thanks, > Biao /'bɪ.aʊ/ > > > > On Mon, 23 Sep 2019 at 19:51, Felipe Gutierrez < > felipe.o.gutier...@gmail.com> wrote: > >> thanks Biao, >> >> I see. To achieve what I want to do I need to work with KeyedStream. I >> downloaded the Flink source code to learn and alter the KeyedStream to my >> needs. I am not sure but it is a lot of work because as far as I understood >> the key-groups have to be predictable [1]. and altering this touches a lot >> of other parts of the source code. >> >> However, If I guarantee that they (key-groups) are predictable, I will be >> able to rebalance, rescale, .... the keys to other worker-nodes. >> >> [1] >> https://flink.apache.org/features/2017/07/04/flink-rescalable-state.html >> >> Thanks, >> Felipe >> *--* >> *-- Felipe Gutierrez* >> >> *-- skype: felipe.o.gutierrez* >> *--* *https://felipeogutierrez.blogspot.com >> <https://felipeogutierrez.blogspot.com>* >> >> >> On Mon, Sep 23, 2019 at 9:51 AM Biao Liu <mmyy1...@gmail.com> wrote: >> >>> Hi Felipe, >>> >>> Flink job graph is DAG based. It seems that you set an "edge property" >>> (partitioner) several times. >>> Flink does not support multiple partitioners on one edge. The later one >>> overrides the priors. That means the "keyBy" overrides the "rebalance" and >>> "partitionByPartial". >>> >>> You could insert some nodes between these partitioners to satisfy your >>> requirement. For example, >>> `sourceDataStream.rebalance().map(...).keyby(0).sum(1).print();` >>> >>> Thanks, >>> Biao /'bɪ.aʊ/ >>> >>> >>> >>> On Thu, 19 Sep 2019 at 16:49, Felipe Gutierrez < >>> felipe.o.gutier...@gmail.com> wrote: >>> >>>> I am executing a data stream application which uses rebalance. >>>> Basically I am counting words using "src -> split -> >>>> physicalPartitionStrategy -> keyBy -> sum -> print". I am running 3 >>>> examples, one without physical partition strategy, one with rebalance >>>> strategy [1], and one with partial partition strategy from [2]. >>>> I know that the keyBy operator actually kills what rebalance is doing >>>> because it splits the stream by key and if I have a stream with skewed key, >>>> one parallel instance of the operator after the keyBy will be overloaded. >>>> However, I was expecting that *before the keyBy* I would have a >>>> balanced stream, which is not happening. >>>> >>>> Basically, I want to see the difference in records/sec between >>>> operators when I use rebalance or any other physical partition strategy. >>>> However, when I found no difference in the records/sec metrics of all >>>> operators when I am running 3 different physical partition strategies. >>>> Screenshots of Prometheus+Grafana are attached. >>>> >>>> Maybe I am measuring the wrong operator, or maybe I am not using the >>>> rebalance in the right way, or I am not doing a good use case to test the >>>> rebalance transformation. >>>> I am also testing a different physical partition to later try to >>>> implement the issue "FLINK-1725 New Partitioner for better load balancing >>>> for skewed data" [2]. I am not sure, but I guess that all physical >>>> partition strategies have to be implemented on a KeyedStream. >>>> >>>> DataStream<String> text = env.addSource(new WordSource()); >>>> // split lines in strings >>>> DataStream<Tuple2<String, Integer>> tokenizer = text.flatMap(new >>>> Tokenizer()); >>>> // choose a partitioning strategy >>>> DataStream<Tuple2<String, Integer>> partitionedStream = tokenizer); >>>> DataStream<Tuple2<String, Integer>> partitionedStream = >>>> tokenizer.rebalance(); >>>> DataStream<Tuple2<String, Integer>> partitionedStream = >>>> tokenizer.partitionByPartial(0); >>>> // count >>>> partitionedStream.keyBy(0).sum(1).print(); >>>> >>>> [1] >>>> https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/#physical-partitioning >>>> [2] https://issues.apache.org/jira/browse/FLINK-1725 >>>> >>>> thanks, >>>> Felipe >>>> >>>> *--* >>>> *-- Felipe Gutierrez* >>>> >>>> *-- skype: felipe.o.gutierrez* >>>> *--* *https://felipeogutierrez.blogspot.com >>>> <https://felipeogutierrez.blogspot.com>* >>>> >>>
