You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Eugenio Marotti <in...@gmail.com> on 2023/07/01 14:36:11 UTC
Average on sliding window
Hi everyone,
I’m trying to calculate an average with a sliding window. Here’s the code I’m using. First of all I receive a series of events from a Kafka topic. I declared a watermark on the ‘data_fine’ attribute.
final TableDescriptor filteredPhasesDurationsTableDescriptor = TableDescriptor.forConnector("upsert-kafka")
.schema(Schema.newBuilder()
.column("id_fascicolo", DataTypes.BIGINT().notNull())
.column("nrg", DataTypes.STRING())
.column("giudice", DataTypes.STRING())
.column("oggetto", DataTypes.STRING())
.column("codice_oggetto", DataTypes.STRING())
.column("ufficio", DataTypes.STRING())
.column("sezione", DataTypes.STRING())
.column("fase", DataTypes.STRING().notNull())
.column("durata", DataTypes.BIGINT())
.column("data_inizio", DataTypes.TIMESTAMP_LTZ(3))
.column("data_fine", DataTypes.TIMESTAMP_LTZ(3))
.watermark("data_fine", "data_fine - INTERVAL '5' SECOND")
.primaryKey("id_fascicolo", "fase")
.build())
.option(KafkaConnectorOptions.TOPIC, List.of("sicid.processor.filtered-phases-durations"))
.option(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS, KAFKA_HOST)
.option(KafkaConnectorOptions.KEY_FORMAT, "json")
.option(KafkaConnectorOptions.VALUE_FORMAT, "json")
.build();
tEnv.createTable("FilteredPhasesDurations_Kafka", filteredPhasesDurationsTableDescriptor);
Table filteredPhasesDurationsTable = tEnv.from("FilteredPhasesDurations_Kafka”);
After receiving the events I compute the average using a sliding window like this:
Table averageJudgeByPhase = filteredPhasesDurationsTable
.window(Slide.over(lit(15).days())
.every(lit(1).days())
.on($("data_fine"))
.as("w"))
.groupBy(
$("giudice"),
$("fase"),
$("w")
)
.select(
$("giudice"),
$("fase"),
$("durata").avg().as("mediaMobileGiudicePerFase"),
$("w").end().as("data_fine")
);
averageJudgeByPhase.execute().print();
But I get no results.
These are some test events in the Kafka topic. The data_fine attribute has a span of more than 15 days so why the average is not computed?
+----+----------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+----------------------+-------------------------+-------------------------+
| op | id_fascicolo | nrg | giudice | oggetto | codice_oggetto | ufficio | sezione | fase | durata | data_inizio | data_fine |
+----+----------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+----------------------+-------------------------+-------------------------+
| +I | 9 | 301 | giudice01 | Usucapione | 131002 | Benevento | 01 | FASE_INTRODUTTIVA | 14 | 2023-04-16 02:00:00.000 | 2023-04-30 02:00:00.000 |
| +I | 10 | 302 | giudice01 | Usucapione | 131002 | Benevento | 01 | FASE_INTRODUTTIVA | 15 | 2023-04-30 02:00:00.000 | 2023-05-15 02:00:00.000 |
| +I | 8 | 300 | giudice01 | Usucapione | 131002 | Benevento | 01 | FASE_INTRODUTTIVA | 14 | 2023-04-01 02:00:00.000 | 2023-04-15 02:00:00.000 |
Thanks,
Eugenio
Re: Average on sliding window
Posted by Alexey Novakov via user <us...@flink.apache.org>.
Hi Eugenio,
I think it is due to window completion which will be complete once your
watermarked field on the event advances 15 days interval since the first
received event time.
Please also check this default trigger behavior here:
https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/datastream/operators/windows/#default-triggers-of-windowassigners
Best regards,
Alexey
On Sat, Jul 1, 2023 at 4:37 PM Eugenio Marotti <
ing.eugenio.marotti@gmail.com> wrote:
> Hi everyone,
>
> I’m trying to calculate an average with a sliding window. Here’s the code
> I’m using. First of all I receive a series of events from a Kafka topic. I
> declared a watermark on the ‘data_fine’ attribute.
>
> final TableDescriptor filteredPhasesDurationsTableDescriptor =
> TableDescriptor.forConnector("upsert-kafka")
> .schema(Schema.newBuilder()
> .column("id_fascicolo", DataTypes.BIGINT().notNull())
> .column("nrg", DataTypes.STRING())
> .column("giudice", DataTypes.STRING())
> .column("oggetto", DataTypes.STRING())
> .column("codice_oggetto", DataTypes.STRING())
> .column("ufficio", DataTypes.STRING())
> .column("sezione", DataTypes.STRING())
> .column("fase", DataTypes.STRING().notNull())
> .column("durata", DataTypes.BIGINT())
> .column("data_inizio", DataTypes.TIMESTAMP_LTZ(3))
> .column("data_fine", DataTypes.TIMESTAMP_LTZ(3))
> .watermark("data_fine", "data_fine - INTERVAL '5' SECOND")
> .primaryKey("id_fascicolo", "fase")
> .build())
> .option(KafkaConnectorOptions.TOPIC, List.of(
> "sicid.processor.filtered-phases-durations"))
> .option(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS, KAFKA_HOST)
> .option(KafkaConnectorOptions.KEY_FORMAT, "json")
> .option(KafkaConnectorOptions.VALUE_FORMAT, "json")
> .build();
> tEnv.createTable("FilteredPhasesDurations_Kafka",
> filteredPhasesDurationsTableDescriptor);
> Table filteredPhasesDurationsTable = tEnv.from(
> "FilteredPhasesDurations_Kafka”);
>
> After receiving the events I compute the average using a sliding window
> like this:
>
> Table averageJudgeByPhase = filteredPhasesDurationsTable
> .window(Slide.over(lit(15).days())
> .every(lit(1).days())
> .on($("data_fine"))
> .as("w"))
> .groupBy(
> $("giudice"),
> $("fase"),
> $("w")
> )
> .select(
> $("giudice"),
> $("fase"),
> $("durata").avg().as("mediaMobileGiudicePerFase"),
> $("w").end().as("data_fine")
> );
>
> averageJudgeByPhase.execute().print();
>
> But I get no results.
> These are some test events in the Kafka topic. The data_fine attribute has
> a span of more than 15 days so why the average is not computed?
>
>
> +----+----------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+----------------------+-------------------------+-------------------------+
> | op | id_fascicolo | nrg |
> giudice | oggetto |
> codice_oggetto | ufficio |
> sezione | fase | durata |
> data_inizio | data_fine |
>
> +----+----------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+----------------------+-------------------------+-------------------------+
> | +I | 9 | 301 |
> giudice01 | Usucapione |
> 131002 | Benevento | 01
> | FASE_INTRODUTTIVA | 14 | 2023-04-16
> 02:00:00.000 | 2023-04-30 02:00:00.000 |
> | +I | 10 | 302 |
> giudice01 | Usucapione |
> 131002 | Benevento | 01
> | FASE_INTRODUTTIVA | 15 | 2023-04-30
> 02:00:00.000 | 2023-05-15 02:00:00.000 |
> | +I | 8 | 300 |
> giudice01 | Usucapione |
> 131002 | Benevento | 01
> | FASE_INTRODUTTIVA | 14 | 2023-04-01
> 02:00:00.000 | 2023-04-15 02:00:00.000 |
>
>
> Thanks,
> Eugenio
>