Réacteur de tâches¶

Mise en œuvre de la tâche worker¶

La tâche worker est chargée d’effectuer une performance attribuée par le répartiteur, telle que l’extraction et l’ingestion de certaines données. La seule partie obligatoire est d’avoir une méthode de tâches worker spécifique qui initie le travail. Cette méthode doit pouvoir être appelée à partir de la procédure Snowpark, renvoyer une chaîne et contenir les paramètres suivants :

• session - Un objet de session Snowpark

• worker_id - Numéro, identifiant unique de la tâche worker

• task_reactor_schema - Nom du schéma dans lequel les objets Task Reactor sont créés. Il peut être utilisé comme nom d’instance de Task Reactor.

La tâche worker est chargée d’exécuter la tâche attribuée par le répartiteur, par exemple d’extraire et d’ingérer des données spécifiques. Nous vous recommandons d’utiliser les classes Java (com.snowflake.connectors.sdk.taskreactor.worker.IngestionWorker et com.snowflake.connectors.sdk.taskreactor.ingestion.Ingestion) ou, pour les tâches plus simples (com.snowflake.connectors.sdk.taskreactor.worker.SimpleTaskWorker et com.snowflake.connectors.sdk.taskreactor.ingestion.SimpleTask), toutefois, votre tâche worker peut être créée dans n’importe quel langage de programmation pris en charge pour l’écriture de gestionnaires (handlers) de procédures stockées.

Exemple de méthode de tâches worker Java :

public static String executeWork(Session session, int workerId, String taskReactorSchema) {
  FakeIngestion fakeIngestion = new FakeIngestion(session, workerId);
  WorkerId workerIdentifier = new WorkerId(workerId);
  Identifier schemaIdentifier = Identifier.fromWithAutoQuoting(taskReactorSchema);
  try {
    IngestionWorker.from(session, fakeIngestion, workerIdentifier, schemaIdentifier).run();
  } catch (WorkerException e) {
    // handle the exception...
    throw new RuntimeException(e);
  }
  return "Worker procedure executed.";
}

Avec une méthode de tâches worker déjà créée, l’utilisateur doit l’intégrer dans CONNECTOR.WORKER_PROCEDURE. La procédure doit appeler sa propre méthode de tâches worker. Elle doit être créée dans le schéma de votre application, renvoyer une STRING et contenir les paramètres suivants :

• worker_id - nombre

• task_reactor_schema - chaîne

Un exemple de procédure, appelant l’implémentation Java de la tâche worker :

CREATE OR REPLACE PROCEDURE CONNECTOR.WORKER_PROCEDURE(worker_id number, task_reactor_schema string)
    RETURNS STRING
    LANGUAGE JAVA
    RUNTIME_VERSION = '11'
    PACKAGES = ('com.snowflake:snowpark:1.11.0', 'com.snowflake:telemetry:0.0.1')
    IMPORTS = ('@jars/myconnector-1.0.0.jar')
    HANDLER = 'com.snowflake.myconnector.WorkerImpl.executeWork';

La bibliothèque de télémétrie est nécessaire pour collecter les métriques qui sont enregistrées dans la table d’événements.

Mise en œuvre du sélecteur¶

Le rôle du sélecteur est de décider quelles tâches en attente doivent être traitées par Task Reactor. Comme pour l’implémentation de la tâche worker, elle peut être créée dans n’importe quel langage pris en charge par Snowpark. Le sélecteur de tâches peut être mis en œuvre sous la forme d’une procédure de base de données ou d’une vue de base de données. Le sélecteur (procédure ou vue) doit être transmis comme argument dans la procédure TASK_REACTOR.CREATE_NEW_INSTANCE.

La procédure doit pouvoir être appelée à partir d’une procédure Snowpark, renvoyer une chaîne et contenir les paramètres suivants :

• session - Session Snowpark

• queueItems - String[] (un tableau de chaînes JSON individuelles, chacune décrivant un seul QueueItem)

Exemple de méthode de sélection Java :

public static String selectWork(Session session, String[] queueItems) {
  Variant[] sorted =
      Arrays.stream(queueItems)
          .map(Variant::new)
          .filter(
              queueItem ->
                  !queueItem.asMap().get("resourceId").asString().equals("filter-out-resource"))
          .sorted(comparing(queueItem -> queueItem.asMap().get("resourceId").asString()))
          .toArray(Variant[]::new);
  return new Variant(sorted).asJsonString();
}

Au lieu de la méthode du sélecteur, il est toujours possible de créer une vue qui filtrera et triera les tâches de la file d’attente existante. Le répartiteur peut récupérer les nouvelles tâches dans la vue nouvellement créée à l’aide d’une requête donnée en exemple :

CREATE VIEW CONNECTOR_SCHEMA.WORK_SELECTOR_VIEW AS SELECT * FROM TASK_REACTOR.QUEUE ORDER BY RESOURCE_ID;

Avec une méthode de sélection déjà créée, l’utilisateur doit l’intégrer dans CONNECTOR.WORK_SELECTOR. La procédure doit appeler votre méthode obligatoire de sélection du travail. Elle doit être créée dans le schéma de votre application, renvoyer un ARRAY et contenir le paramètre suivant :

• work_items - array

Un exemple de procédure, appelant l’implémentation Java du sélecteur de travail :

CREATE OR REPLACE PROCEDURE CONNECTOR.WORK_SELECTOR(work_items array)
    RETURNS ARRAY
    LANGUAGE JAVA
    RUNTIME_VERSION = '11'
    PACKAGES = ('com.snowflake:snowpark:1.11.0')
    IMPORTS = ('@jars/myconnector-1.0.0.jar')
    HANDLER = 'com.snowflake.myconnector.WorkSelector.selectWork';

Mise en œuvre du sélecteur expiré¶

Le rôle du sélecteur expiré est de décider quels éléments de la file d’attente doivent être retirés de la file d’attente de Task Reactor. Il peut être nécessaire de supprimer des éléments parce que le sélecteur ne peut jamais atteindre certains éléments et que ceux-ci resteraient éternellement dans la file d’attente. En outre, certains éléments qui attendent dans la file d’attente peuvent avoir été créés longtemps auparavant et il n’est plus utile de les traiter. Le sélecteur expiré peut être mis en œuvre sous la forme d’une vue de la base de données. La vue du sélecteur doit être transmise en tant qu’argument dans la procédure TASK_REACTOR.CREATE_NEW_INSTANCE. S’il n’est pas nécessaire de retirer des éléments de la file d’attente, l’implémentation par défaut peut être utilisée TASK_REACTOR.EMPTY_EXPIRED_WORK_SELECTOR.

En utilisant la requête suivante, il est possible de créer une vue avec sélecteur expiré qui sélectionne les éléments qui ont été créés il y a plus de 3 jours :

CREATE VIEW CONNECTOR_SCHEMA.EXPIRED_WORK_SELECTOR_VIEW
    AS SELECT * FROM TASK_REACTOR.QUEUE q
        WHERE DATEDIFF(day, q.timestamp, sysdate()) > 3;

Intégrer des objets d’instance¶

Le TASK_REACTOR.CREATE_INSTANCE_OBJECTS permet à l’utilisateur de configurer toutes les instances ensemble avant d’initialiser les instances créées. La procédure ne peut être exécutée qu’une seule fois par schéma, de sorte que tout appel ultérieur n’entraînera aucune modification. Nous vous recommandons de placer l’appel d’initialisation dans le fichier setup.sql, afin d’éviter que la procédure ne soit exécutée plusieurs fois ou qu’elle ne soit pas appelée du tout.

Paramètres requis :

• instance_schema_name VARCHAR - Un schéma unique par instance qui stocke les objets de la base de données sur lesquels l’instance travaille.

• worker_procedure_name VARCHAR - Nom de la procédure de la tâche worker décrite dans la partie Worker Implementation.

• work_selector_type VARCHAR - Valeurs indiquant si les nouvelles tâches doivent utiliser la vue ou la procédure. Valeurs possibles : VIEW, PROCEDURE.

• work_selector_name VARCHAR - Nom de la procédure/vue de sélection décrite dans la partie Selector Implementation.

Paramètres facultatifs :

• expired_work_selector_name VARCHAR - Nom de la vue du sélecteur expirée décrite dans la partie Expired Selector Implementation. Si la valeur n’est pas fournie, TASK_REACTOR.EMPTY_EXPIRED_WORK_SELECTOR est utilisé comme implémentation par défaut qui ne renvoie rien.