タスクリアクター¶

ワーカーの実装¶

ワーカーは、特定のデータの取得や取り込みなど、ディスパッチャーによって割り当てられたタスクを実行する責任を負います。唯一必須なのは、ジョブを開始する特定のワーカーメソッドを持つことです。このメソッドは、Snowparkプロシージャから呼び出し可能で、文字列を返し、次のパラメーターを含んでいる必要があります。

• session - Snowparkセッションオブジェクト

• worker_id - 数値、ユニークワーカーID

• task_reactor_schema - タスクリアクターオブジェクトが作成されるスキーマ名。タスクリアクターのインスタンス名として使用できます。

ワーカーは、特定のデータの取得や取り込みなど、ディスパッチャーによって割り当てられたタスクを実行する責任を負います。（com.snowflake.connectors.sdk.taskreactor.worker.IngestionWorker および com.snowflake.connectors.sdk.taskreactor.ingestion.Ingestion）Javaクラスを使用するか、より単純なタスク（com.snowflake.connectors.sdk.taskreactor.worker.SimpleTaskWorker および com.snowflake.connectors.sdk.taskreactor.ingestion.SimpleTask）を使用することをお勧めしますが、ストアドプロシージャのハンドラーの作成がサポートされている任意のプログラミング言語でワーカーを作成できます。

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.";
}

すでに作成されたワーカーメソッドを使用する場合、ユーザーはそれを CONNECTOR.WORKER_PROCEDURE に統合する必要があります。プロシージャは独自のワーカーメソッドを呼び出す必要があります。これはアプリケーションスキーマ内に作成され、 STRING を返し、次のパラメーターを含める必要があります。

• worker_id - 数値

• task_reactor_schema - 文字列

ワーカーのJava実装を呼び出すプロシージャの例:

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

イベントテーブルにログ記録されるメトリクスを収集するには、テレメトリライブラリが必要です。

セレクターの実装¶

セレクターの役割は、キューに入れられたタスクのうち、どのタスクをタスクリアクターで処理するかを決定することです。ワーカーの実装と同様に、Snowparkでサポートされている任意の言語で作成できます。タスクセレクターは、データベースプロシージャまたはデータベースビューとして実装できます。セレクター（プロシージャまたはビュー）は、 TASK_REACTOR.CREATE_NEW_INSTANCE プロシージャの引数として渡す必要があります。

プロシージャはSnowparkプロシージャから呼び出し可能で、文字列を返し、次のパラメーターを含んでいる必要があります。

• session - Snowparkセッション

• queueItems - String[]（個々の JSON 文字列の配列。各文字列は単一の QueueItem を表します）

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();
}

セレクターメソッドの代わりに、既存のキューからタスクをフィルタリングおよびソートするビューを作成することもできます。ディスパッチャーは、次のサンプルクエリを使用して、新しく作成されたビューから新しいタスクを取得できます。

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

ユーザーは、すでに作成されているセレクターメソッドを使用して CONNECTOR.WORK_SELECTOR に統合する必要があります。このプロシージャーは、必須のワークセレクターメソッドを呼び出す必要があります。これはアプリケーションスキーマ内に作成され、 ARRAY を返し、次のパラメーターを含める必要があります。

• work_items - array

ワークセレクターのJava実装を呼び出すプロシージャの例：

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

期限切れセレクターの実装¶

期限切れセレクターの役割は、タスクリアクターのキューから削除するキューアイテムを決定することです。セレクターが一部のアイテムに到達できず、これらのアイテムがキュー内に永久に残るため、アイテムを削除する必要がある場合があります。さらに、キューで待機している一部のアイテムは、かなり前に作成された可能性があり、それ以上処理が必要ないものもあります。期限切れセレクターは、データベースビューとして実装できます。セレクタービューは、 TASK_REACTOR.CREATE_NEW_INSTANCE プロシージャの引数として渡す必要があります。キューからアイテムを削除する必要がない場合は、デフォルトの実装である TASK_REACTOR.EMPTY_EXPIRED_WORK_SELECTOR を使用できます。

次のクエリを使用すると、3日以上前に作成されたアイテムを選択する期限切れセレクタービューを作成できます。

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

インスタンスオブジェクトの統合¶

TASK_REACTOR.CREATE_INSTANCE_OBJECTS を使用すると、作成されたインスタンスを初期化する前に、すべてのインスタンスをまとめて構成できます。このプロシージャはスキーマごとに1回だけ実行できるため、今後の呼び出しでは変更は行われません。プロシージャが何度も実行されたり、まったく呼び出されなかったりするのを防ぐために、 setup.sql ファイルに初期化コールを置くことをお勧めします。

必須パラメーター:

• instance_schema_name VARCHAR - インスタンスが動作するデータベースオブジェクトを格納する、インスタンスごとに一意のスキーマ。

• worker_procedure_name VARCHAR - Worker Implementation の部分に記述されているワーカープロシージャの名前。

• work_selector_type VARCHAR - - 新しいタスクで、ビューまたはプロシージャのどちらを使用するかを指定する値。可能な値: VIEW、 PROCEDURE。

• work_selector_name VARCHAR - Selector Implementation - 部分に記述されているセレクタープロシージャまたはビューの名前。

オプションのパラメーター:

• expired_work_selector_name VARCHAR - Expired Selector Implementation - 部分に記述されている期限切れのセレクタービューの名前。値が提供されていない場合、 TASK_REACTOR.EMPTY_EXPIRED_WORK_SELECTOR は何も返さないデフォルトの実装として使用されます。