How can I improve this Flink application for "Distinct Count of elements" in the data stream?

[Felipe Gutierrez]

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe

[Rong Rong]

Hi Felipe,

there are multiple ways to do DISTINCT COUNT in Table/SQL API. In fact there's already a thread going on recently [1] 

Based on the description you provided, it seems like it might be a better API level to use.

To answer your question,

- You should be able to use other TimeCharacteristic. You might want to try WindowProcessFunction and see if this fits your use case.

- Not sure I fully understand the question, your keyed by should be done on your distinct key (or a combo key) and if you do keyby correctly then yes all msg with same key is processed by the same TM thread.

--

Rong

[1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/count-DISTINCT-in-flink-SQL-td28061.html

On Tue, Jun 11, 2019 at 1:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe

[Felipe Gutierrez]

Hi Rong,

thanks for your answer. If I understood well, the option will be to use ProcessFunction [1] since it has the method onTimer(). But not the ProcessWindowFunction [2], because it does not have the method onTimer(). I will need this method to call Collector<OUT> out.collect(...) from the onTImer() method in order to emit a single value of my Distinct Count function.

Is that reasonable what I am saying?

[1] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/datastream/DataStream.html
[2] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.html

Kind Regards,

Felipe

On Wed, Jun 12, 2019 at 3:41 AM Rong Rong <[hidden email]> wrote:

Hi Felipe,

there are multiple ways to do DISTINCT COUNT in Table/SQL API. In fact there's already a thread going on recently [1] 

Based on the description you provided, it seems like it might be a better API level to use.

To answer your question,

- You should be able to use other TimeCharacteristic. You might want to try WindowProcessFunction and see if this fits your use case.

- Not sure I fully understand the question, your keyed by should be done on your distinct key (or a combo key) and if you do keyby correctly then yes all msg with same key is processed by the same TM thread.

--

Rong

[1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/count-DISTINCT-in-flink-SQL-td28061.html

On Tue, Jun 11, 2019 at 1:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe

[Felipe Gutierrez]

Hi Rong, I implemented my solution using a ProcessingWindow with timeWindow and a ReduceFunction with timeWindowAll [1]. So for the first window I use parallelism and the second window I use to merge everything on the Reducer. I guess it is the best approach to do DistinctCount. Would you suggest some improvements?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountProcessTimeWindowSocket.java

Thanks!

On Wed, Jun 12, 2019 at 9:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong,

thanks for your answer. If I understood well, the option will be to use ProcessFunction [1] since it has the method onTimer(). But not the ProcessWindowFunction [2], because it does not have the method onTimer(). I will need this method to call Collector<OUT> out.collect(...) from the onTImer() method in order to emit a single value of my Distinct Count function.

Is that reasonable what I am saying?

[1] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/datastream/DataStream.html
[2] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.html

Kind Regards,

Felipe

On Wed, Jun 12, 2019 at 3:41 AM Rong Rong <[hidden email]> wrote:

Hi Felipe,

there are multiple ways to do DISTINCT COUNT in Table/SQL API. In fact there's already a thread going on recently [1] 

Based on the description you provided, it seems like it might be a better API level to use.

To answer your question,

- You should be able to use other TimeCharacteristic. You might want to try WindowProcessFunction and see if this fits your use case.

- Not sure I fully understand the question, your keyed by should be done on your distinct key (or a combo key) and if you do keyby correctly then yes all msg with same key is processed by the same TM thread.

--

Rong

[1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/count-DISTINCT-in-flink-SQL-td28061.html

On Tue, Jun 11, 2019 at 1:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe

[Hequn Cheng]

Hi Felipe,

From your code, I think you want to get the "count distinct" result instead of the "distinct count". They contain a different meaning. 

To improve the performance, you can replace your DistinctProcessWindowFunction to a DistinctProcessReduceFunction. A ReduceFunction can aggregate the elements of a window incrementally, while for ProcessWindowFunction, elements cannot be incrementally aggregated but instead need to be buffered internally until the window is considered ready for processing.

> Flink does not have a built-in operator which does this computation.

Flink does have built-in operators to solve your problem. You can use Table API & SQL to solve your problem. The code looks like:

public static void main(String[] args) throws Exception {
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

   DataStream ds = env.socketTextStream("localhost", 9000);
   tableEnv.registerDataStream("sourceTable", ds, "line, proctime.proctime");

   SplitTableFunction splitFunc = new SplitTableFunction();
   tableEnv.registerFunction("splitFunc", splitFunc);
   Table result = tableEnv.scan("sourceTable")
         .joinLateral("splitFunc(line) as word")
         .window(Tumble.over("5.seconds").on("proctime").as("w"))
         .groupBy("w")
         .select("count.distinct(word), collect.distinct(word)");

   tableEnv.toAppendStream(result, Row.class).print();
   env.execute();
}

Detail code can be found here[1].

At the same time, you can perform two-stage window to improve the performance, i.e., the first window aggregate is used to distinct words and the second window used to get the final results.

Document about Table API and SQL can be found here[2][3].

Best, Hequn

[1] https://github.com/hequn8128/flink/commit/b4676a5730cecabe2931b9e628aaebd7729beab2

[2] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/tableApi.html

[3] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/sql.html

On Wed, Jun 12, 2019 at 9:19 PM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong, I implemented my solution using a ProcessingWindow with timeWindow and a ReduceFunction with timeWindowAll [1]. So for the first window I use parallelism and the second window I use to merge everything on the Reducer. I guess it is the best approach to do DistinctCount. Would you suggest some improvements?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountProcessTimeWindowSocket.java

Thanks!

On Wed, Jun 12, 2019 at 9:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong,

thanks for your answer. If I understood well, the option will be to use ProcessFunction [1] since it has the method onTimer(). But not the ProcessWindowFunction [2], because it does not have the method onTimer(). I will need this method to call Collector<OUT> out.collect(...) from the onTImer() method in order to emit a single value of my Distinct Count function.

Is that reasonable what I am saying?

[1] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/datastream/DataStream.html
[2] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.html

Kind Regards,

Felipe

On Wed, Jun 12, 2019 at 3:41 AM Rong Rong <[hidden email]> wrote:

Hi Felipe,

there are multiple ways to do DISTINCT COUNT in Table/SQL API. In fact there's already a thread going on recently [1] 

Based on the description you provided, it seems like it might be a better API level to use.

To answer your question,

- You should be able to use other TimeCharacteristic. You might want to try WindowProcessFunction and see if this fits your use case.

- Not sure I fully understand the question, your keyed by should be done on your distinct key (or a combo key) and if you do keyby correctly then yes all msg with same key is processed by the same TM thread.

--

Rong

[1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/count-DISTINCT-in-flink-SQL-td28061.html

On Tue, Jun 11, 2019 at 1:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe

[Felipe Gutierrez]

Hi Hequn,

indeed the ReduceFunction is better than the ProcessWindowFunction. I replaced and could check the improvement performance [1]. Thanks for that!

I will try a distinct count with the Table API.

The question that I am facing is that I want to use a HyperLogLog on a UDF for DataStream. Thus I will be able to have an approximate distinct count inside a window, like I did here [2]. After having my UDF I want to have my own operator which process this approximation of distinct count. So I am not sure with I can implement my own operator for the TableAPI. Can I?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountReduceWindowSocket.java

[2] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordHLLKeyedProcessWindowSocket.java

Thanks!

Felipe

On Thu, Jun 13, 2019 at 8:10 AM Hequn Cheng <[hidden email]> wrote:

Hi Felipe,

From your code, I think you want to get the "count distinct" result instead of the "distinct count". They contain a different meaning. 

To improve the performance, you can replace your DistinctProcessWindowFunction to a DistinctProcessReduceFunction. A ReduceFunction can aggregate the elements of a window incrementally, while for ProcessWindowFunction, elements cannot be incrementally aggregated but instead need to be buffered internally until the window is considered ready for processing.

> Flink does not have a built-in operator which does this computation.

Flink does have built-in operators to solve your problem. You can use Table API & SQL to solve your problem. The code looks like:

public static void main(String[] args) throws Exception {
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

   DataStream ds = env.socketTextStream("localhost", 9000);
   tableEnv.registerDataStream("sourceTable", ds, "line, proctime.proctime");

   SplitTableFunction splitFunc = new SplitTableFunction();
   tableEnv.registerFunction("splitFunc", splitFunc);
   Table result = tableEnv.scan("sourceTable")
         .joinLateral("splitFunc(line) as word")
         .window(Tumble.over("5.seconds").on("proctime").as("w"))
         .groupBy("w")
         .select("count.distinct(word), collect.distinct(word)");

   tableEnv.toAppendStream(result, Row.class).print();
   env.execute();
}

Detail code can be found here[1].

At the same time, you can perform two-stage window to improve the performance, i.e., the first window aggregate is used to distinct words and the second window used to get the final results.

Document about Table API and SQL can be found here[2][3].

Best, Hequn

[1] https://github.com/hequn8128/flink/commit/b4676a5730cecabe2931b9e628aaebd7729beab2

[2] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/tableApi.html

[3] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/sql.html

On Wed, Jun 12, 2019 at 9:19 PM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong, I implemented my solution using a ProcessingWindow with timeWindow and a ReduceFunction with timeWindowAll [1]. So for the first window I use parallelism and the second window I use to merge everything on the Reducer. I guess it is the best approach to do DistinctCount. Would you suggest some improvements?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountProcessTimeWindowSocket.java

Thanks!

On Wed, Jun 12, 2019 at 9:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong,

thanks for your answer. If I understood well, the option will be to use ProcessFunction [1] since it has the method onTimer(). But not the ProcessWindowFunction [2], because it does not have the method onTimer(). I will need this method to call Collector<OUT> out.collect(...) from the onTImer() method in order to emit a single value of my Distinct Count function.

Is that reasonable what I am saying?

[1] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/datastream/DataStream.html
[2] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.html

Kind Regards,

Felipe

On Wed, Jun 12, 2019 at 3:41 AM Rong Rong <[hidden email]> wrote:

Hi Felipe,

there are multiple ways to do DISTINCT COUNT in Table/SQL API. In fact there's already a thread going on recently [1] 

Based on the description you provided, it seems like it might be a better API level to use.

To answer your question,

- You should be able to use other TimeCharacteristic. You might want to try WindowProcessFunction and see if this fits your use case.

- Not sure I fully understand the question, your keyed by should be done on your distinct key (or a combo key) and if you do keyby correctly then yes all msg with same key is processed by the same TM thread.

--

Rong

[1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/count-DISTINCT-in-flink-SQL-td28061.html

On Tue, Jun 11, 2019 at 1:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe

[Rong Rong]

Hi Felipe,

Hequn is right. The problem you are facing is better using TableAPI level code instead of dealing with in DataStream. You will have more Flink library support to achieve your goal.

In addition, Flink TableAPI also support UserDefineAggregateFunction [1] to achieve your hyperLogLog based approximation. In fact the interface is similar to the ones in DataStream API [2].

--

Rong

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/udfs.html#aggregation-functions

[2] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/operators/windows.html#aggregatefunction

On Thu, Jun 13, 2019 at 8:55 AM Felipe Gutierrez <[hidden email]> wrote:

Hi Hequn,

indeed the ReduceFunction is better than the ProcessWindowFunction. I replaced and could check the improvement performance [1]. Thanks for that!

I will try a distinct count with the Table API.

The question that I am facing is that I want to use a HyperLogLog on a UDF for DataStream. Thus I will be able to have an approximate distinct count inside a window, like I did here [2]. After having my UDF I want to have my own operator which process this approximation of distinct count. So I am not sure with I can implement my own operator for the TableAPI. Can I?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountReduceWindowSocket.java

[2] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordHLLKeyedProcessWindowSocket.java

Thanks!

Felipe

On Thu, Jun 13, 2019 at 8:10 AM Hequn Cheng <[hidden email]> wrote:

Hi Felipe,

From your code, I think you want to get the "count distinct" result instead of the "distinct count". They contain a different meaning. 

To improve the performance, you can replace your DistinctProcessWindowFunction to a DistinctProcessReduceFunction. A ReduceFunction can aggregate the elements of a window incrementally, while for ProcessWindowFunction, elements cannot be incrementally aggregated but instead need to be buffered internally until the window is considered ready for processing.

> Flink does not have a built-in operator which does this computation.

Flink does have built-in operators to solve your problem. You can use Table API & SQL to solve your problem. The code looks like:

public static void main(String[] args) throws Exception {
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

   DataStream ds = env.socketTextStream("localhost", 9000);
   tableEnv.registerDataStream("sourceTable", ds, "line, proctime.proctime");

   SplitTableFunction splitFunc = new SplitTableFunction();
   tableEnv.registerFunction("splitFunc", splitFunc);
   Table result = tableEnv.scan("sourceTable")
         .joinLateral("splitFunc(line) as word")
         .window(Tumble.over("5.seconds").on("proctime").as("w"))
         .groupBy("w")
         .select("count.distinct(word), collect.distinct(word)");

   tableEnv.toAppendStream(result, Row.class).print();
   env.execute();
}

Detail code can be found here[1].

At the same time, you can perform two-stage window to improve the performance, i.e., the first window aggregate is used to distinct words and the second window used to get the final results.

Document about Table API and SQL can be found here[2][3].

Best, Hequn

[1] https://github.com/hequn8128/flink/commit/b4676a5730cecabe2931b9e628aaebd7729beab2

[2] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/tableApi.html

[3] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/sql.html

On Wed, Jun 12, 2019 at 9:19 PM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong, I implemented my solution using a ProcessingWindow with timeWindow and a ReduceFunction with timeWindowAll [1]. So for the first window I use parallelism and the second window I use to merge everything on the Reducer. I guess it is the best approach to do DistinctCount. Would you suggest some improvements?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountProcessTimeWindowSocket.java

Thanks!

On Wed, Jun 12, 2019 at 9:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong,

thanks for your answer. If I understood well, the option will be to use ProcessFunction [1] since it has the method onTimer(). But not the ProcessWindowFunction [2], because it does not have the method onTimer(). I will need this method to call Collector<OUT> out.collect(...) from the onTImer() method in order to emit a single value of my Distinct Count function.

Is that reasonable what I am saying?

[1] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/datastream/DataStream.html
[2] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.html

Kind Regards,

Felipe

On Wed, Jun 12, 2019 at 3:41 AM Rong Rong <[hidden email]> wrote:

Hi Felipe,

there are multiple ways to do DISTINCT COUNT in Table/SQL API. In fact there's already a thread going on recently [1] 

Based on the description you provided, it seems like it might be a better API level to use.

To answer your question,

- You should be able to use other TimeCharacteristic. You might want to try WindowProcessFunction and see if this fits your use case.

- Not sure I fully understand the question, your keyed by should be done on your distinct key (or a combo key) and if you do keyby correctly then yes all msg with same key is processed by the same TM thread.

--

Rong

[1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/count-DISTINCT-in-flink-SQL-td28061.html

On Tue, Jun 11, 2019 at 1:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe

[Felipe Gutierrez]

humm.. it seems that it is my turn to implement all this stuff using Table API.

Thanks Rong!

On Thu, Jun 13, 2019 at 6:00 PM Rong Rong <[hidden email]> wrote:

Hi Felipe,

Hequn is right. The problem you are facing is better using TableAPI level code instead of dealing with in DataStream. You will have more Flink library support to achieve your goal.

In addition, Flink TableAPI also support UserDefineAggregateFunction [1] to achieve your hyperLogLog based approximation. In fact the interface is similar to the ones in DataStream API [2].

--

Rong

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/udfs.html#aggregation-functions

[2] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/operators/windows.html#aggregatefunction

On Thu, Jun 13, 2019 at 8:55 AM Felipe Gutierrez <[hidden email]> wrote:

Hi Hequn,

indeed the ReduceFunction is better than the ProcessWindowFunction. I replaced and could check the improvement performance [1]. Thanks for that!

I will try a distinct count with the Table API.

The question that I am facing is that I want to use a HyperLogLog on a UDF for DataStream. Thus I will be able to have an approximate distinct count inside a window, like I did here [2]. After having my UDF I want to have my own operator which process this approximation of distinct count. So I am not sure with I can implement my own operator for the TableAPI. Can I?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountReduceWindowSocket.java

[2] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordHLLKeyedProcessWindowSocket.java

Thanks!

Felipe

On Thu, Jun 13, 2019 at 8:10 AM Hequn Cheng <[hidden email]> wrote:

Hi Felipe,

From your code, I think you want to get the "count distinct" result instead of the "distinct count". They contain a different meaning. 

To improve the performance, you can replace your DistinctProcessWindowFunction to a DistinctProcessReduceFunction. A ReduceFunction can aggregate the elements of a window incrementally, while for ProcessWindowFunction, elements cannot be incrementally aggregated but instead need to be buffered internally until the window is considered ready for processing.

> Flink does not have a built-in operator which does this computation.

Flink does have built-in operators to solve your problem. You can use Table API & SQL to solve your problem. The code looks like:

public static void main(String[] args) throws Exception {
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

   DataStream ds = env.socketTextStream("localhost", 9000);
   tableEnv.registerDataStream("sourceTable", ds, "line, proctime.proctime");

   SplitTableFunction splitFunc = new SplitTableFunction();
   tableEnv.registerFunction("splitFunc", splitFunc);
   Table result = tableEnv.scan("sourceTable")
         .joinLateral("splitFunc(line) as word")
         .window(Tumble.over("5.seconds").on("proctime").as("w"))
         .groupBy("w")
         .select("count.distinct(word), collect.distinct(word)");

   tableEnv.toAppendStream(result, Row.class).print();
   env.execute();
}

Detail code can be found here[1].

At the same time, you can perform two-stage window to improve the performance, i.e., the first window aggregate is used to distinct words and the second window used to get the final results.

Document about Table API and SQL can be found here[2][3].

Best, Hequn

[1] https://github.com/hequn8128/flink/commit/b4676a5730cecabe2931b9e628aaebd7729beab2

[2] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/tableApi.html

[3] https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/table/sql.html

On Wed, Jun 12, 2019 at 9:19 PM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong, I implemented my solution using a ProcessingWindow with timeWindow and a ReduceFunction with timeWindowAll [1]. So for the first window I use parallelism and the second window I use to merge everything on the Reducer. I guess it is the best approach to do DistinctCount. Would you suggest some improvements?

[1] https://github.com/felipegutierrez/explore-flink/blob/master/src/main/java/org/sense/flink/examples/stream/WordDistinctCountProcessTimeWindowSocket.java

Thanks!

On Wed, Jun 12, 2019 at 9:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi Rong,

thanks for your answer. If I understood well, the option will be to use ProcessFunction [1] since it has the method onTimer(). But not the ProcessWindowFunction [2], because it does not have the method onTimer(). I will need this method to call Collector<OUT> out.collect(...) from the onTImer() method in order to emit a single value of my Distinct Count function.

Is that reasonable what I am saying?

[1] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/datastream/DataStream.html
[2] https://ci.apache.org/projects/flink/flink-docs-master/api/java/index.html?org/apache/flink/streaming/api/functions/windowing/ProcessWindowFunction.html

Kind Regards,

Felipe

On Wed, Jun 12, 2019 at 3:41 AM Rong Rong <[hidden email]> wrote:

Hi Felipe,

there are multiple ways to do DISTINCT COUNT in Table/SQL API. In fact there's already a thread going on recently [1] 

Based on the description you provided, it seems like it might be a better API level to use.

To answer your question,

- You should be able to use other TimeCharacteristic. You might want to try WindowProcessFunction and see if this fits your use case.

- Not sure I fully understand the question, your keyed by should be done on your distinct key (or a combo key) and if you do keyby correctly then yes all msg with same key is processed by the same TM thread.

--

Rong

[1] http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/count-DISTINCT-in-flink-SQL-td28061.html

On Tue, Jun 11, 2019 at 1:27 AM Felipe Gutierrez <[hidden email]> wrote:

Hi all,

I have implemented a Flink data stream application to compute distinct count of words. Flink does not have a built-in operator which does this computation. I used KeyedProcessFunction and I am saving the state on a ValueState descriptor.
Could someone check if my implementation is the best way of doing it? Here is my solution: https://stackoverflow.com/questions/56524962/how-can-i-improve-my-count-distinct-for-data-stream-implementation-in-flink/56539296#56539296

I have some points that I could not understand better:
- I only could use TimeCharacteristic.IngestionTime.
- I split the words using "Tuple2<Integer, String>(0, word)", so I will have always the same key (0). As I understand, all the events will be processed on the same TaskManager which will not achieve parallelism if I am in a cluster.

Kind Regards,
Felipe