Dans le traitement de flux avec Apache Flink, la gestion du temps est cruciale pour définir des fenêtres sur les données. Flink propose trois caractéristiques temporelles : le temps d'événement (EventTime), le temps d'ingestion (IngestionTime) et le temps de traitement (ProcessingTime). Chacune influence la manière dont les fenêtres sont créées et appliquées.

• Le temps d'événement (EventTime) se base sur l'horodatage inclus dans les messages eux-mêmes, ce qui permet de traiter les données en fonction de leur moment de génération, indépendamment de leur ordre d'arrivée.
• Le temps d'ingestion (IngestionTime) utilise l'horodatage au moment où les données entrent dans la source Flink, offrant une gestion plus simple mais moins flexible.
• Le temps de traitement (ProcessingTime) correspond au temps de l'opérateur lors de l'exécution, ce qui le rend plus rapide mais moins précis pour les événements hors ordre.

Utilisation avencée avec le temps d'événement

Lorsque vous basculer les fenêtres sur le temps d'événement, les messages doivent contenir un horodatage événementiel. Par exemple, considérons des données de ventes avec des horodatages en millisecondes. Voici un jeu de données simulé :

1609459200000,vendeurA,produitX,150.0
1609459201000,vendeurB,produitX,250.0
1609459202000,vendeurA,produitY,300.0
1609459203000,vendeurB,produitX,450.0
1609459204000,vendeurA,produitX,500.0

Dans un premier code, on configure l'environnement pour utiliser le temps d'événement et on définit une fenêtre glissante. Cependant, ce code de base ignore les problèmes de retard réseau, ce qui peut entraîner des résultats incohérents.

object FenetreEvenementSimple {
  def main(args: Array[String]): Unit = {
    val environnement = StreamExecutionEnvironment.getExecutionEnvironment
    environnement.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
    val fluxSource = environnement.socketTextStream("localhost", 9999)
    case class EnregistrementVente(timestamp: Long, vendeur: String, produit: String, montant: Double)
    
    val fluxTransforme = fluxSource.map(ligne => {
      val champs = ligne.split(",")
      EnregistrementVente(champs(0).toLong, champs(1), champs(2), champs(3).toDouble)
    })
    
    val fluxAvecTemps = fluxTransforme.assignAscendingTimestamps(_.timestamp)
    val fluxParProduit = fluxAvecTemps.keyBy(_.produit)
    val fenetreProduit = fluxParProduit.timeWindow(Time.seconds(5))
    val resultats = fenetreProduit.max("montant")
    
    resultats.print()
    environnement.execute()
  }
}

Pour gérer les retards et les désordres, il est essentiel d'ajouter des watermarks. Celles-ci délimitent le progrès du temps et permettent de gérer les données en retard. Voici une implémentation améliorée avec un assigneur de wtaermarks périodiques.

object FenetreEvenementAvecWatermark {
  def main(args: Array[String]): Unit = {
    val environnement = StreamExecutionEnvironment.getExecutionEnvironment
    environnement.setParallelism(1)
    environnement.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
    val fluxSource = environnement.socketTextStream("localhost", 9999)
    case class EnregistrementVente(timestamp: Long, vendeur: String, produit: String, montant: Double)
    
    val fluxDonnees = fluxSource.map(ligne => {
      val champs = ligne.split(",")
      EnregistrementVente(champs(0).toLong, champs(1), champs(2), champs(3).toDouble)
    })
    
    val fluxAvecWatermark = fluxDonnees.assignTimestampsAndWatermarks(
      new AssignerWithPeriodicWatermarks[EnregistrementVente] {
        var maxTimestampCourant: Long = 0L
        val delaiMaxHorsOrdre = 3000L // 3 secondes de retard maximum
        
        override def getCurrentWatermark: Watermark = {
          new Watermark(maxTimestampCourant - delaiMaxHorsOrdre)
        }
        
        override def extractTimestamp(element: EnregistrementVente, previousTimestamp: Long): Long = {
          val ts = element.timestamp
          maxTimestampCourant = Math.max(ts, maxTimestampCourant)
          ts
        }
      }
    )
    
    val fluxParProduit = fluxAvecWatermark.keyBy(_.produit)
    val fenetreProduit = fluxParProduit.timeWindow(Time.seconds(5))
    val resultats = fenetreProduit.apply(new FonctionFenetrePersonnalisee)
    
    resultats.print()
    environnement.execute()
  }
}

class FonctionFenetrePersonnalisee extends WindowFunction[EnregistrementVente, EnregistrementVente, String, TimeWindow] {
  override def apply(cle: String, fenetre: TimeWindow, elements: Iterable[EnregistrementVente], collecteur: Collector[EnregistrementVente]): Unit = {
    val tableauTrie = elements.toArray.sortBy(_.montant).reverse.take(2) // Exemple : garder les deux montants les plus élevés
    tableauTrie.foreach(collecteur.collect)
  }
}

Utilisation avec le temps de traitement

Le temps de traitement ne nécessite pas de watermarks, car il se base sur le temps réel de l'opérateur. Cela simplifie le code, mais les résultats peuvent varier selon l'ordre d'arrivée des données.

object FenetreTraitementSimple {
  def main(args: Array[String]): Unit = {
    val environnement = StreamExecutionEnvironment.getExecutionEnvironment
    environnement.setParallelism(2)
    environnement.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
    
    val fluxSource = environnement.socketTextStream("localhost", 9999)
    case class EnregistrementVente(timestamp: Long, vendeur: String, produit: String, montant: Double)
    
    val fluxDonnees = fluxSource.map(ligne => {
      val champs = ligne.split(",")
      EnregistrementVente(champs(0).toLong, champs(1), champs(2), champs(3).toDouble)
    })
    
    val fluxParProduit = fluxDonnees.keyBy(_.produit)
    val fenetreProduit = fluxParProduit.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
    val resultats = fenetreProduit.min("montant") // Exemple : trouver le montant minimum
    
    resultats.print()
    environnement.execute()
  }
}

Dans cet exemple, la fenêtre se déclenche en fonction du temps de traitement, ce qui est utile pour des applications où la fraîcheur des données est piroritaire sur l'ordre temporel précis.