# Tumbling Window giving incorrect aggregated results

**URL:** <https://forum.confluent.io/t/tumbling-window-giving-incorrect-aggregated-results/7341>\
**Category:** ksqlDB\
**Created:** [4 March 2023 07:09 UTC](https://forum.confluent.io/t/tumbling-window-giving-incorrect-aggregated-results/7341 "2023-03-04T07:09:20Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![hnabhijith](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/hnabhijith/32/2779_2.png) [@hnabhijith](https://forum.confluent.io/u/hnabhijith)\
**Post date:** [4 March 2023 07:09 UTC](https://forum.confluent.io/t/tumbling-window-giving-incorrect-aggregated-results/7341/1 "2023-03-04T07:09:21Z")

</div>

KSQL Windowing (Tumbling Window) is used for aggregating incoming stream of data. 5 min window is defined with a grace period of 30 mins.  
The KSQL scripts used to achieve the use case is attached. It is observed that on running aggregation continuously for long periods the aggregated values seem to be calculated incorrectly for few windows randomly.

Data set for reproducing the issue is attached. Data set is prepared/loaded as below:  
10K records are loaded for a duration of 40 Mins. Loading interval between each file is 20 seconds. So for every 5 mins a total of 150000 records are loaded. The data is prepared in such a way that records loaded in 5mins interval (150000) all fall under a 5 min interval (Ex: 10:00 - 10:05). The dimension used for aggregation has a granularity of 10 unique values. So when a 5 min window is aggregated expected record count per 5 min window is 10.

On running the aggregation we are seeing that aggregated values for the columns are coming incorrectly for few partitions randomly. The aggregated values seems to be lesser than the actual values expected.

**KSQL Scripts for the aggregation flow**  
CREATE STREAM RT\_SMS\_RAW\_1  
(EVENT\_TIME STRING, IMSI STRING,MSISDN STRING,IMEI STRING,MCC STRING,MNC STRING,SAC\_ID BIGINT,CELL\_ID BIGINT,LAC BIGINT,CLEAR\_CODE\_ID BIGINT,DELIVERY\_TIME STRING,RADIO\_ACCESS\_INFO BIGINT,SMS\_TYPE BIGINT,SMS\_LENGTH BIGINT,ORIG\_CALLING\_NUM STRING,CALLED\_NUMBER STRING,INCOMING\_TIME STRING,SOURCE\_ID STRING,SMS\_SERV\_CENTER\_ADDR\_NUMBER STRING,SMS\_MSC\_TYPE BIGINT,SMS\_BSC\_ID STRING,SMS\_RNC\_ID STRING,SMS\_MSC\_ID STRING,STAGE\_TYPE STRING,MVNO\_ID BIGINT,SMS\_START\_TIME STRING,SMS\_RP\_ORIGINATING\_ADDRESS STRING,SMS\_RP\_ORIGINATING\_ADDRESS\_NOA BIGINT,SMS\_RP\_ORIGINATING\_ADDRESS\_NP BIGINT,SMS\_RP\_DESTINATION\_ADDRESS STRING,SMS\_RP\_DESTINATION\_ADDRESS\_NOA BIGINT,SMS\_RP\_DESTINATION\_ADDRESS\_NP BIGINT,SMS\_TP\_ORIGINATING\_ADDRESS STRING,SMS\_TP\_ORIGINATING\_ADDRESS\_NOA BIGINT,SMS\_TP\_ORIGINATING\_ADDRESS\_NP BIGINT,SMS\_TP\_DESTINATION\_ADDRESS STRING,SMS\_TP\_DESTINATION\_ADDRESS\_NOA BIGINT,SMS\_TP\_DESTINATION\_ADDRESS\_NP BIGINT,SMS\_MESSAGE\_REFERENCE\_NUMBER BIGINT,SMS\_IE\_IDENTIFIER STRING,SMS\_CONCATENATED\_REFERENCE BIGINT,SMS\_CONCATENATED\_MAX BIGINT,SMS\_CONCATENATED\_SEQ BIGINT,SMS\_MORE\_MESSAGES\_TO\_SEND BIGINT,SMS\_RP\_CAUSE BIGINT,SMS\_CP\_CAUSE BIGINT,SMS\_TP\_CAUSE BIGINT,SMS\_SMSC\_TIME STRING,SMS\_RP\_ACK\_TIME STRING,PCSCF\_ID STRING,ENB\_ID STRING,UUID STRING,PROCESS\_SEQUENCE\_NUMBER STRING,SIP\_START\_TIME STRING,SIP\_END\_TIME STRING,CALL\_ID STRING,PROBE\_SESSION\_ID STRING,REPORT\_REASON STRING,USER\_AGENT STRING, EXPIRATION BIGINT,MCC\_MNC\_IMSI STRING,DEVICE\_TAC STRING, RETRANSMISSION\_COUNT BIGINT,LAST\_200\_OK\_TIME STRING,FROM\_URI\_USERNAME STRING,TO\_URI\_USERNAME STRING,FROM\_URI\_HOST STRING,TO\_URI\_HOST STRING,FROM\_URI\_TELEPHONE\_SUBSCRIBER STRING,TO\_URI\_TELEPHONE\_SUBSCRIBER STRING,LAST\_REASON\_PHRASE STRING, SETUP\_DURATION BIGINT, STATUS BIGINT,SOURCE\_IP\_ADDRESS STRING,SOURCE\_PORT BIGINT,DESTINATION\_IP\_ADDRESS STRING,DESTINATION\_PORT BIGINT,CSEQ BIGINT,VERSION STRING,P\_PREFERRED\_USER\_ID STRING,VIA\_BRANCH STRING,ROUTE STRING,INTERFACE\_CODE BIGINT,GTP\_CAUSE\_CODE BIGINT,GTP\_RECORD\_TYPE BIGINT,GTP\_SESSION\_ID STRING,GTP\_SIGNALLING\_START\_TIME STRING,GTP\_SOURCE\_IP\_ADDRESS STRING,GTP\_DESTINATION\_IP\_ADDRESS STRING,ACCESS\_TYPE\_FIRST\_PANI\_A STRING,ACCESS\_TYPE\_FIRST\_PANI\_B STRING,FIRST\_TRYING\_TIME STRING,MO\_MT\_TYPE BIGINT,P\_ACCESS\_NETWORK\_INFO\_FIRST\_A STRING,P\_ACCESS\_NETWORK\_INFO\_FIRST\_B STRING,P\_ASSERTED\_ID STRING,P\_CHARGING\_VECTOR\_ICID STRING,P\_CHARGING\_VECTOR\_ORIGIN\_IOI STRING,P\_CHARGING\_VECTOR\_TERM\_IOI STRING,P\_CHARGING\_VECTOR\_TRANSIT\_IOI STRING,FIRST\_REASON STRING,REASON\_TEXT STRING,SIP\_REASON\_HEADER\_CODE BIGINT,VLAN\_PCP BIGINT,VLAN\_CFI BIGINT,VLAN\_VID BIGINT,WLAN\_NODE\_ID\_FIRST\_PANI\_A STRING,WLAN\_NODE\_ID\_FIRST\_PANI\_B STRING)  
WITH (KAFKA\_TOPIC=‘sms.data’,VALUE\_FORMAT=‘AVRO’,TIMESTAMP=‘EVENT\_TIME’,TIMESTAMP\_FORMAT=‘yyyy-MM-dd HH:mm:ss.SSS’);

CREATE STREAM RT\_SMS\_DENORM\_1 WITH (KAFKA\_TOPIC=‘rt\_sms\_denorm\_1\_t’,TIMESTAMP=‘EVENT\_TIME’,TIMESTAMP\_FORMAT=‘yyyy-MM-dd HH:mm:ss.SSS’,PARTITIONS=25) AS SELECT FORMAT\_TIMESTAMP(CONVERT\_TZ(PARSE\_TIMESTAMP(s.EVENT\_TIME, ‘yyyy-MM-dd HH:mm:ss.SSS’),‘LOCAL\_TZ’,‘UTC’),‘yyyy-MM-dd HH:mm:ss.SSS’) AS EVENT\_TIME, S.CLEAR\_CODE\_ID AS CLEAR\_CODE\_ID, S.PCSCF\_ID AS PCSCF\_ID, S.SMS\_SERV\_CENTER\_ADDR\_NUMBER AS SMS\_SERV\_CENTER\_ADDR\_NUMBER, S.SMS\_TYPE AS SMS\_TYPE, S.RADIO\_ACCESS\_INFO AS RADIO\_ACCESS\_INFO, (CASE WHEN e.CLEAR\_CODE\_STATUS = ‘fail’ THEN 1 ELSE 0 END) AS NUMBER\_OF\_SMS\_FAILURES, (CASE WHEN s.SMS\_TYPE=0 AND e.CLEAR\_CODE\_STATUS = ‘success’ THEN (UNIX\_TIMESTAMP(PARSE\_TIMESTAMP(DELIVERY\_TIME, ‘yyyy-MM-dd HH:mm:ss.SSS’))-UNIX\_TIMESTAMP(PARSE\_TIMESTAMP(INCOMING\_TIME, ‘yyyy-MM-dd HH:mm:ss.SSS’)))/1000 ELSE NULL END) AS MO\_DELIVERY\_TIME, (CASE WHEN s.SMS\_TYPE=2 AND e.CLEAR\_CODE\_STATUS = ‘success’ THEN (UNIX\_TIMESTAMP(PARSE\_TIMESTAMP(DELIVERY\_TIME, ‘yyyy-MM-dd HH:mm:ss.SSS’))-UNIX\_TIMESTAMP(PARSE\_TIMESTAMP(INCOMING\_TIME, ‘yyyy-MM-dd HH:mm:ss.SSS’)))/1000 ELSE NULL END) AS MT\_DELIVERY\_TIME, (CASE WHEN s.SMS\_TYPE=0 AND e.CLEAR\_CODE\_STATUS = ‘success’ THEN 1 ELSE 0 END) AS MO\_TOTAL\_SUCCESS\_SMS, (CASE WHEN s.SMS\_TYPE=2 AND e.CLEAR\_CODE\_STATUS = ‘success’ THEN 1 ELSE 0 END) AS MT\_TOTAL\_SUCCESS\_SMS FROM RT\_SMS\_RAW\_1 S LEFT OUTER JOIN RT\_DIM\_CLEAR\_CODE\_1 E ON S.CLEAR\_CODE\_ID=E.CLEAR\_CODE\_ID EMIT CHANGES;

CREATE TABLE RT\_SMS\_PCSCF\_4G\_1\_5MIN WITH (KEY\_FORMAT=‘AVRO’,KAFKA\_TOPIC=‘rt\_sms\_pcscf\_4g\_1\_t’,VALUE\_FORMAT=‘AVRO’,PARTITIONS=25) AS SELECT CAST(WINDOWSTART AS STRING) AS WINDOW\_START\_TIME, ‘P-CSCF’ AS NETWORK\_ELEMENT\_TYPE, CASE WHEN PCSCF\_ID IS NULL THEN ‘NULL IN SOURCE’ ELSE PCSCF\_ID END AS NETWORK\_ELEMENT\_ID, AS\_VALUE(‘P-CSCF’) AS NE\_TYPE, AS\_VALUE(CASE WHEN PCSCF\_ID IS NULL THEN ‘NULL IN SOURCE’ ELSE PCSCF\_ID END) AS NE\_ID, SUM(NUMBER\_OF\_SMS\_FAILURES) AS NUMBER\_OF\_SMS\_FAILURES, COUNT(\*) AS NUMBER\_OF\_SMS, ROUND(SUM(MO\_DELIVERY\_TIME)) AS MO\_DELIVERY\_TIME, ROUND(SUM(MT\_DELIVERY\_TIME)) AS MT\_DELIVERY\_TIME, SUM(MO\_TOTAL\_SUCCESS\_SMS) AS MO\_TOTAL\_SUCCESS\_SMS, SUM(MT\_TOTAL\_SUCCESS\_SMS) AS MT\_TOTAL\_SUCCESS\_SMS FROM RT\_SMS\_DENORM\_1 WINDOW TUMBLING(SIZE 5 MINUTE,RETENTION 6 HOURS, GRACE PERIOD 35 MINUTES) WHERE SMS\_TYPE IN (0, 2) AND RADIO\_ACCESS\_INFO = 5 GROUP BY ‘P-CSCF’, CASE WHEN PCSCF\_ID IS NULL THEN ‘NULL IN SOURCE’ ELSE PCSCF\_ID END EMIT CHANGES;

CREATE STREAM RT\_SMS\_PCSCF\_TS\_4G\_1\_5MIN ( WINDOW\_START\_TIME STRING, NE\_TYPE STRING, NE\_ID STRING, NUMBER\_OF\_SMS\_FAILURES BIGINT, NUMBER\_OF\_SMS BIGINT, MO\_DELIVERY\_TIME BIGINT, MT\_DELIVERY\_TIME BIGINT, MO\_TOTAL\_SUCCESS\_SMS BIGINT, MT\_TOTAL\_SUCCESS\_SMS BIGINT ) WITH(KAFKA\_TOPIC=‘rt\_sms\_pcscf\_4g\_1\_t’,VALUE\_FORMAT=‘AVRO’,PARTITIONS=25);

create stream RT\_SMS\_PCSCF\_REKEY\_4G\_1\_5MIN WITH (KEY\_FORMAT=‘KAFKA’,kafka\_topic=‘rt\_sms\_pcscf\_4g\_rekey\_1\_t’,value\_format=‘JSON’,partitions=25) as select concat (WINDOW\_START\_TIME + ‘:’ + NE\_ID) as COMPOSITE\_KEY, FORMAT\_TIMESTAMP(FROM\_UNIXTIME(CAST(WINDOW\_START\_TIME AS BIGINT)),‘yyyy-MM-dd HH:mm:ss.SSS’,‘LOCAL\_TZ’) as report\_time, ne\_type, ne\_id, number\_of\_sms\_failures, number\_of\_sms, mo\_delivery\_time, mt\_delivery\_time, mo\_total\_success\_sms, mt\_total\_success\_sms from RT\_SMS\_PCSCF\_TS\_4G\_1\_5MIN partition by CONCAT(WINDOW\_START\_TIME + ‘:’ + NE\_ID);

CREATE TABLE RT\_SMS\_SMSC\_4G\_1\_5MIN WITH (KEY\_FORMAT=‘AVRO’,KAFKA\_TOPIC=‘rt\_sms\_smsc\_4g\_1\_t’,VALUE\_FORMAT=‘AVRO’,PARTITIONS=25) AS SELECT CAST(WINDOWSTART AS STRING) AS WINDOW\_START\_TIME, ‘SMSC’ AS NETWORK\_ELEMENT\_TYPE, CASE WHEN SMS\_SERV\_CENTER\_ADDR\_NUMBER IS NULL THEN ‘NULL IN SOURCE’ ELSE SMS\_SERV\_CENTER\_ADDR\_NUMBER END AS NETWORK\_ELEMENT\_ID, AS\_VALUE(‘SMSC’) AS NE\_TYPE, AS\_VALUE(CASE WHEN SMS\_SERV\_CENTER\_ADDR\_NUMBER IS NULL THEN ‘NULL IN SOURCE’ ELSE SMS\_SERV\_CENTER\_ADDR\_NUMBER END) AS NE\_ID, SUM(NUMBER\_OF\_SMS\_FAILURES) AS NUMBER\_OF\_SMS\_FAILURES, COUNT(\*) AS NUMBER\_OF\_SMS, ROUND(SUM(MO\_DELIVERY\_TIME)) AS MO\_DELIVERY\_TIME, ROUND(SUM(MT\_DELIVERY\_TIME)) AS MT\_DELIVERY\_TIME, SUM(MO\_TOTAL\_SUCCESS\_SMS) AS MO\_TOTAL\_SUCCESS\_SMS, SUM(MT\_TOTAL\_SUCCESS\_SMS) AS MT\_TOTAL\_SUCCESS\_SMS FROM RT\_SMS\_DENORM\_1 WINDOW TUMBLING(SIZE 5 MINUTE,RETENTION 6 HOURS, GRACE PERIOD 35 MINUTES) WHERE SMS\_TYPE IN (0, 2) AND RADIO\_ACCESS\_INFO = 5 GROUP BY ‘SMSC’, CASE WHEN SMS\_SERV\_CENTER\_ADDR\_NUMBER IS NULL THEN ‘NULL IN SOURCE’ ELSE SMS\_SERV\_CENTER\_ADDR\_NUMBER END EMIT CHANGES;

CREATE STREAM RT\_SMS\_SMSC\_TS\_4G\_1\_5MIN ( WINDOW\_START\_TIME STRING, NE\_TYPE STRING, NE\_ID STRING, NUMBER\_OF\_SMS\_FAILURES BIGINT, NUMBER\_OF\_SMS BIGINT, MO\_DELIVERY\_TIME BIGINT, MT\_DELIVERY\_TIME BIGINT, MO\_TOTAL\_SUCCESS\_SMS BIGINT, MT\_TOTAL\_SUCCESS\_SMS BIGINT ) WITH(KAFKA\_TOPIC=‘rt\_sms\_smsc\_4g\_1\_t’,VALUE\_FORMAT=‘AVRO’,PARTITIONS=25);

create stream RT\_SMS\_SMSC\_REKEY\_4G\_1\_5MIN WITH (KEY\_FORMAT=‘KAFKA’,kafka\_topic=‘rt\_sms\_smsc\_4g\_rekey\_1\_t’,value\_format=‘JSON’,partitions=25) as select concat(WINDOW\_START\_TIME + ‘:’ + NE\_ID) as COMPOSITE\_KEY, FORMAT\_TIMESTAMP(FROM\_UNIXTIME(CAST(WINDOW\_START\_TIME AS BIGINT)),‘yyyy-MM-dd HH:mm:ss.SSS’,‘LOCAL\_TZ’) as report\_time, ne\_type, ne\_id, number\_of\_sms\_failures, number\_of\_sms, mo\_delivery\_time , mt\_delivery\_time , mo\_total\_success\_sms, mt\_total\_success\_sms from RT\_SMS\_SMSC\_TS\_4G\_1\_5MIN partition by concat (WINDOW\_START\_TIME + ‘:’ + NE\_ID);

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [3 April 2023 07:10 UTC](https://forum.confluent.io/t/tumbling-window-giving-incorrect-aggregated-results/7341/2 "2023-04-03T07:10:10Z")

</div>

Closed on 3 April 2023 07:10 UTC

This topic was automatically closed after 30 days. New replies are no longer allowed.