[
https://issues.apache.org/jira/browse/BEAM-5690?focusedWorklogId=325076&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-325076
]
ASF GitHub Bot logged work on BEAM-5690:
----------------------------------------
Author: ASF GitHub Bot
Created on: 08/Oct/19 13:51
Start Date: 08/Oct/19 13:51
Worklog Time Spent: 10m
Work Description: echauchot commented on pull request #9567: [BEAM-5690]
Fix Zero value issue with GroupByKey/CountByKey in SparkRunner
URL: https://github.com/apache/beam/pull/9567#discussion_r332476939
##########
File path:
runners/spark/src/main/java/org/apache/beam/runners/spark/stateful/SparkGroupAlsoByWindowViaWindowSet.java
##########
@@ -338,6 +339,18 @@ public void outputWindowedValue(
outputHolder.getWindowedValues();
if (!outputs.isEmpty() || !stateInternals.getState().isEmpty()) {
+ Collection<TimerInternals.TimerData> filteredTimers =
+ timerInternals.getTimers().stream()
+ .filter(
+ timer ->
+ timer
+ .getTimestamp()
+ .plus(windowingStrategy.getAllowedLateness())
+
.isBefore(timerInternals.currentInputWatermarkTime()))
+ .collect(Collectors.toList());
+
+ filteredTimers.forEach(timerInternals::deleteTimer);
+
Review comment:
Logic with watermarks is good. I first thought we should put that it the
core utility method `LateDataUtils.dropExpiredWindows` but this method drops
elements based on their windows and watermark. I know think that dropping the
internals timers more belongs to this (dedicated to timers) part of the code.
but you should wrap this code into a method such as dropExpiredTimers and maybe
put it in a core utility such as `LateDataUtils` because it is a common matter
of all the runners
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 325076)
Time Spent: 1h 20m (was: 1h 10m)
> Issue with GroupByKey in BeamSql using SparkRunner
> --------------------------------------------------
>
> Key: BEAM-5690
> URL: https://issues.apache.org/jira/browse/BEAM-5690
> Project: Beam
> Issue Type: Task
> Components: runner-spark
> Reporter: Kenneth Knowles
> Priority: Major
> Time Spent: 1h 20m
> Remaining Estimate: 0h
>
> Reported on user@
> {quote}We are trying to setup a pipeline with using BeamSql and the trigger
> used is default (AfterWatermark crosses the window).
> Below is the pipeline:
>
> KafkaSource (KafkaIO)
> ---> Windowing (FixedWindow 1min)
> ---> BeamSql
> ---> KafkaSink (KafkaIO)
>
> We are using Spark Runner for this.
> The BeamSql query is:
> {code}select Col3, count(*) as count_col1 from PCOLLECTION GROUP BY Col3{code}
> We are grouping by Col3 which is a string. It can hold values string[0-9].
>
> The records are getting emitted out at 1 min to kafka sink, but the output
> record in kafka is not as expected.
> Below is the output observed: (WST and WET are indicators for window start
> time and window end time)
> {code}
> {"count_col1":1,"Col3":"string5","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":3,"Col3":"string7","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":2,"Col3":"string8","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":1,"Col3":"string2","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":1,"Col3":"string6","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":0,"Col3":"string6","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":0,"Col3":"string6","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":0,"Col3":"string6","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":0,"Col3":"string6","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":0,"Col3":"string6","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0000 +0000"}
> {"count_col1":0,"Col3":"string6","WST":"2018-10-09 09-55-00 0000
> +0000","WET":"2018-10-09 09-56-00 0}
> {code}
> {quote}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
