Peut-on utiliser un opérateur d'entrée au milieu d'un DAG dans Apache Apex

Question

Tous les exemples d'Apex disent que le premier opérateur du DAG doit être un opérateur d'entrée. Est-ce que cet opérateur peut apparaître quelque part au milieu du DAG ?

Considérons un cas dans lequel j'ai des données à récupérer de la base de données, basées sur des données qui viennent d'être traitées par un opérateur précédent, cela signifierait qu'un opérateur d'entrée apparaîtra quelque part au milieu du DAG.

Selon la définition d'un opérateur d'entrée, il s'agit d'un opérateur qui n'a pas de flux d'entrée. Mais il fait aussi le travail de récupération des données si un connecteur est utilisé. Est-ce que cela fonctionnera si je récupère des données quelque part au milieu d'un DAG ?

Answer 1

Ceci est un cas d'utilisation intéressant. Vous devriez être capable d'étendre un opérateur d'entrée (par exemple JdbcInputOperator si vous voulez lire depuis une base de données) et ajouter un port d'entrée. Ce port d'entrée reçoit des données (tuples) d'un autre opérateur de votre DAG et met à jour la clause "where" du JdbcInputOperator afin qu'il lise les données en fonction de cela. J'espère que c'est ce que vous cherchiez.

Answer 2

Oui, c'est possible. Vous pouvez étendre un InputOperator existant et y ajouter des InputPort(s). Dans ce cas, la plateforme Apex traitera votre opérateur comme un opérateur générique et n'appellera pas InputOperator.emitTuples(). Il sera de la responsabilité de votre opérateur étendu d'appeler super.emitTuples() ou d'émettre directement sur le(s) port(s) de sortie.

Answer 3

Non, un opérateur d'entrée ne peut pas être utilisé entre le DAG. Comme vous l'avez déjà souligné, étant donné qu'il n'y a pas de flux d'entrée, vous ne pourrez pas obtenir de données à partir de l'opérateur précédent pour les utiliser avec cet opérateur.

Pour l'exemple que vous avez pointé, il serait préférable d'écrire votre propre opérateur générique avec un flux d'entrée qui a en fait des fonctionnalités similaires à l'opérateur d'entrée, où il peut lire des données à partir d'une source externe en fonction des données dans le flux d'entrée.

Aussi, juste une remarque : Si la requête est trop lourde, il est préférable d'avoir un thread asynchrone pour interroger la base de données. Ce thread peut écrire des données dans une file d'attente à partir de laquelle le thread principal peut lire les enregistrements et les émettre sur le flux de sortie. Cela garantira que le thread principal de l'opérateur n'est pas bloqué et que l'opérateur ne échouera pas.