Lecture de deux flux (principal et configs) en séquentiel dans Flink

J'ai deux flux, l'un est le flux principal, disons dans l'exemple de la détection de fraude, j'ai un flux de transactions, puis j'ai un deuxième flux qui est configs, dans notre exemple, ce sont des règles. Je connecte donc le flux principal au flux de configuration afin d'effectuer le traitement. Mais lorsque flink démarre pour la première fois et que nous ajoutons un travail, il commence à consommer des transactions et des flux de configuration parallèles et lorsque vous souhaitez traiter la transaction, il constate parfois qu'il n'y a pas de configuration et que nous devons envoyer la transaction à la file d'attente des lettres mortes. Cependant, ce que je veux réaliser, c'est que s'il existe une configuration brevetée que je pourrais obtenir un peu plus tard, je veux d'abord obtenir cette configuration, puis obtenir une transaction afin de la traiter plutôt que de l'envoyer dans la file d'attente des lettres mortes. J'ai la même clé pour les transactions et les configurations.

Pour faire court, existe-t-il un moyen de dire à flink lorsque le premier travail démarre, essayez de consommer un flux jusqu'à ce qu'il n'y ait pas de nouvelle valeur, puis commencez à traiter le flux principal? Comment puis-je les rendre séquentiels?


Solution du problème

La méthode recommandée consiste à connecter les 2 flux et à appliquer un RichCoFlatMap qui vous permettra de mettre en mémoire tampon les événements mainpendant que vous attendez de recevoir les événements de configuration.

Consultez cette section utile des didacticiels Flink. Le tout dernier paragraphe décrit en fait votre problème.

It is important to recognize that you have no control over the order in which the flatMap1 and flatMap2 callbacks are called. These two input streams are racing against each other, and the Flink runtime will do what it wants to regarding consuming events from one stream or the other. In cases where timing and/or ordering matter, you may find it necessary to buffer events in managed Flink state until your application is ready to process them. (Note: if you are truly desperate, it is possible to exert some limited control over the order in which a two-input operator consumes its inputs by using a custom Operator that implements the InputSelectable interface.

Donc, en un mot, vous devriez connecter vos 2 flux et avoir une sorte de ListState où vous pouvez "tamponner" vos éléments principaux en attendant de recevoir les règles. Lorsque vous recevez un élément du flux de configuration, vous vérifiez si vous aviez des éléments en attente "en attente" pour cette configuration dans votre ListState (votre tampon). Si vous le faites, vous pouvez ensuite traiter ces éléments et les émettre via le collecteur de votre flatmap.

Commentaires

Enregistrer un commentaire

Posts les plus consultés de ce blog

Erreur Symfony : "Une exception a été levée lors du rendu d'un modèle"

j'essaie d'inclure <link href="{{ asset('css/mystyle.css') }}" rel="stylesheet"/> dans mon fichier twig et le rendre. Mais cela me donne une exception PHP non interceptée Twig_Error_Runtime : "Une exception a été levée lors du rendu d'un modèle ("Le fichier manifeste de l'actif "/../public/build/manifest.json" n'existe pas.") ça marche quand j'utilise link href="css/mystyle.css" rel="stylesheet"/> . Voici mon contrôleur: <?php namespace App\Controller; use Symfony\Component\Routing\Annotation\Route; use Symfony\Bundle\FrameworkBundle\Controller\AbstractController; use Symfony\Component\HttpFoundation\Response; class WelcomeController extends AbstractController { /** * @Route("/", name="welcome") */ public function index() { return $this->render('welcome/index.html.twig'); } } Voici mon modèle de brindille : <!doctype html> ...
Lire la suite

Détecter les appuis sur les touches fléchées en JavaScript

Comment détecter l'appui sur l'une des touches fléchées ? J'en ai profité pour savoir: function checkKey(e) { var event = window.event? window.event: e; console.log(event.keyCode) } Bien que cela fonctionnait pour toutes les autres touches, ce n'était pas le cas pour les touches fléchées (peut-être parce que le navigateur est censé faire défiler ces touches par défaut). Solution du problème event.key === "Flèche droite"... Plus récent et beaucoup plus propre: utilisez event.key . Plus de codes numériques arbitraires ! Si vous transpilez ou savez que vos utilisateurs utilisent tous des navigateurs modernes, utilisez ceci ! node.addEventListener('keydown', function(event) { const key = event.key; // "ArrowRight", "ArrowLeft", "ArrowUp", or "ArrowDown" }); Gestion détaillée : switch (event.key) { case "ArrowLeft": // Left pressed break; case "ArrowRight": // Right pressed break; c...
Lire la suite

Une chaîne vide donne "Des erreurs ont été détectées dans les arguments de la ligne de commande, veuillez vous assurer que tous les arguments sont correctement définis"

Image
Sur un serveur SQL 2016, j'ai un travail qui appelle un package SSIS. Ce package se trouve dans un projet dans la SSISDB et a des paramètres. L'un de ces paramètres est un type de chaîne vide par défaut. J'ai exécuté le travail avec la valeur vide pour ce paramètre, et il s'est exécuté avec succès. J'ai ensuite ouvert les propriétés du travail, je suis allé à l'étape qui appelle ce package et je suis entré dans la configuration et j'ai donné une valeur à ce paramètre. J'ai exécuté à nouveau le travail et il s'est exécuté avec succès, et la valeur du paramètre a eu l'effet attendu sur les résultats. Maintenant, je veux modifier le travail et redéfinir le paramètre sur une chaîne vide. En répétant ce que j'ai fait ci-dessus, j'ouvre la configuration et supprime complètement la valeur du paramètre. Lorsque j'essaie ensuite d'enregistrer les propriétés de l'étape du travail, j'obtiens une boîte de dialogue d'erreur a...
Lire la suite