Reator de tarefa¶

Implementação do trabalhador¶

O trabalhador é responsável por executar uma tarefa atribuída pelo dispatcher, como extrair e ingerir determinados dados. A única parte obrigatória é ter um método de trabalho específico que inicie o trabalho. Este método deve ser chamado a partir do procedimento Snowpark, retornar uma cadeia de caracteres e conter os seguintes parâmetros:

• session - Objeto de sessão do Snowpark

• worker_id - Número, identificação única do trabalhador

• task_reactor_schema - Nome do esquema onde os objetos do reator de tarefas são criados. Pode ser usado como um nome de instância do reator de tarefas.

O trabalhador é responsável por executar a tarefa atribuída pelo dispatcher, por exemplo, extrair e ingerir dados específicos. Recomendamos usar as classes Java (com.snowflake.connectors.sdk.taskreactor.worker.IngestionWorker e com.snowflake.connectors.sdk.taskreactor.ingestion.Ingestion) ou para tarefas mais simples (com.snowflake.connectors.sdk.taskreactor.worker.SimpleTaskWorker e com.snowflake.connectors.sdk.taskreactor.ingestion.SimpleTask), no entanto, seu trabalhador pode ser criado em qualquer linguagem de programação suportada para escrever manipuladores de procedimentos armazenados.

Exemplo de método do trabalhador de 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.";
}

Com um método do trabalhador já criado, o usuário deve integrá-lo ao CONNECTOR.WORKER_PROCEDURE. O procedimento deve chamar seu próprio método do trabalhador. Ele deve ser criado em seu esquema de aplicativo, retornar uma STRING e conter os seguintes parâmetros:

• worker_id - número

• task_reactor_schema - cadeia de caracteres

Um procedimento de exemplo, chamando a implementação Java do trabalhador:

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';

A biblioteca de telemetria é necessária para coletar métricas que são registradas na tabela de eventos.

Implementação do seletor¶

O trabalho do seletor é decidir quais tarefas enfileiradas devem ser manipuladas pelo reator de tarefas. Semelhante à implementação do trabalhador, ele pode ser criado em qualquer linguagem suportada pelo Snowpark. O seletor de tarefas pode ser implementado como um procedimento de banco de dados ou uma exibição de banco de dados. O seletor (procedimento ou exibição) deve ser passado como argumento no procedimento TASK_REACTOR.CREATE_NEW_INSTANCE.

O procedimento deve ser chamável de um procedimento Snowpark, retornar uma cadeia de caracteres e conter os seguintes parâmetros:

• session - Sessão de Snowpark

• queueItems- Cadeias de caracteres[] (uma matriz de cadeias de caracteres JSON individuais, cada uma descrevendo um único QueueItem)

Exemplo de método do seletor de 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();
}

Em vez do método seletor, ainda é possível criar uma exibição que filtrará e classificará tarefas da fila existente. O dispatcher pode recuperar novas tarefas da exibição recém-criada usando uma consulta de exemplo:

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

Com o método do seletor já criado, o usuário deve integrá-lo ao CONNECTOR.WORK_SELECTOR. O procedimento deve chamar seu método seletor do trabalhador obrigatório. Ele deve ser criado no esquema de seu aplicativo, retornar uma ARRAY, e contém o seguinte parâmetro:

• work_items - array

Um procedimento de exemplo, chamando a implementação Java do seletor do trabalhador:

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';

Implementação do seletor expirado¶

A função do seletor expirado é decidir quais itens na fila devem ser removidos da fila do reator de tarefas. Pode ser necessário remover itens porque o seletor nunca consegue alcançar alguns itens e esses itens permaneceriam na fila para sempre. Além disso, alguns itens que estão esperando na fila podem ser criados muito tempo antes e não faz mais sentido processá-los. O seletor expirado pode ser implementado como uma exibição de banco de dados. A exibição do seletor deve ser passada como um argumento no procedimento TASK_REACTOR.CREATE_NEW_INSTANCE. Se não houver necessidade de remover itens da fila, a implementação padrão pode ser usada TASK_REACTOR.EMPTY_EXPIRED_WORK_SELECTOR.

Usando a consulta a seguir, é possível criar uma exibição de seletor expirado que seleciona os itens que foram criados há mais de 3 dias:

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

Integração de objetos de instância¶

O TASK_REACTOR.CREATE_INSTANCE_OBJECTS permite que o usuário configure todas as instâncias juntas antes de inicializar as instâncias criadas. O procedimento pode ser executado apenas uma vez por esquema, portanto, quaisquer chamadas futuras não afetarão nenhuma alteração. Recomendamos colocar a chamada de inicialização para o arquivo setup.sql, para evitar que o procedimento seja executado várias vezes ou nem seja chamado.

Parâmetros obrigatórios:

• instance_schema_name VARCHAR - Um esquema exclusivo por instância que armazena objetos de banco de dados nos quais a instância trabalha.

• worker_procedure_name VARCHAR - Nome do procedimento do trabalhador descrito na parte Worker Implementation.

• work_selector_type VARCHAR - Valores especificando se novas tarefas devem usar exibição ou procedimento. Possíveis valores: VIEW, PROCEDURE.

• work_selector_name VARCHAR - Nome do procedimento/exibição do seletor descrito na função Selector Implementation.

Parâmetros opcionais:

• expired_work_selector_name VARCHAR - Nome da exibição do seletor expirado descrita na função Expired Selector Implementation. Se o valor não for fornecido, TASK_REACTOR.EMPTY_EXPIRED_WORK_SELECTOR será usado como uma implementação padrão que não retorna nada.