チュートリアル1: Snowpark Container Servicesサービスを作成する¶

紹介¶

共通セットアップ が完了すると、サービスを作成する準備が整います。このチュートリアルでは、入力されたテキストを単純にエコーバックするサービス（echo_service という名前）を作成します。たとえば、入力文字列が「Hello World」の場合、サービスは「I said, Hello World」を返します。

このチュートリアルには2つのパートがあります。

パート1: サービスを作成してテストする。 このチュートリアルで提供されるコードをダウンロードし、ステップバイステップの手順に従います。

1. このチュートリアルのサービスコードをダウンロードします。

2. Snowpark Container Services用のDockerイメージをビルドし、アカウントのリポジトリにイメージをアップロードします。

3. サービスを作成し、サービス仕様ファイルとサービスを実行するコンピューティングプールを提供します。

4. サービスと通信するためのサービス関数を作成します。

5. サービスを使用します。サービスにエコーリクエストを送り、その応答を検証します。

パート2: サービスを理解する。このセクションでは、サービスコードの概要を説明し、さまざまなコンポーネントがどのように連携しているかを明らかにします。

1: サービスコードをダウンロードする¶

Echoサービスを作成するためのコード（Pythonアプリケーション）が提供されます。

1. zipファイル をディレクトリにダウンロードします。

2. ファイルを解凍します。チュートリアルごとに1つのディレクトリが含まれています。 Tutorial-1 ディレクトリには以下のファイルがあります。

   • Dockerfile

   • echo_service.py

   • templates/basic_ui.html

2: イメージをビルドしてアップロードする¶

Snowpark Container Servicesがサポートするlinux/amd64プラットフォーム用のイメージをビルドし、アカウントのイメージリポジトリにイメージをアップロードします（共通セットアップ を参照）。

イメージをビルドしてアップロードする前に、リポジトリに関する情報（リポジトリ URL とレジストリのホスト名）が必要です。詳細については、 レジストリおよびリポジトリ をご参照ください。

リポジトリに関する情報を取得する

1. リポジトリ URL を取得するには、 SHOW IMAGE REPOSITORIES SQL コマンドを実行します。

   SHOW IMAGE REPOSITORIES;
   

   Copy

   • 出力の repository_url 列は、 URL を提供します。以下に例を示します。

     <orgname>-<acctname>.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository
     

   • リポジトリ URL のホスト名はレジストリのホスト名です。以下に例を示します。

     <orgname>-<acctname>.registry.snowflakecomputing.com
     

イメージをビルドし、リポジトリにアップロードする

1. ターミナルウィンドウを開き、解凍したファイルのあるディレクトリに移動します。

2. Dockerイメージをビルドするには、Docker CLI を使用して以下の docker build コマンドを実行します。このコマンドは、イメージのビルドに使用するファイルの PATH として、現在の作業ディレクトリ（.）を指定していることに注意してください。

   docker build --rm --platform linux/amd64 -t <repository_url>/<image_name> .
   

   Copy

   • image_name には、 my_echo_service_image:latest を使用します。

   例

   docker build --rm --platform linux/amd64 -t myorg-myacct.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository/my_echo_service_image:latest .
   

   Copy

3. Snowflakeアカウントのリポジトリにイメージをアップロードします。Dockerがあなたの代わりにリポジトリにイメージをアップロードするには、まずSnowflakeでDockerを認証する必要があります。

   1. SnowflakeレジストリでDockerを認証するには、以下のコマンドを実行します。

      docker login <registry_hostname> -u <username>
      

      Copy

      • username には、Snowflakeのユーザー名を指定します。Dockerは、パスワードの入力を求めるプロンプトを表示します。

   2. イメージをアップロードするには、以下のコマンドを実行します。

      docker push <repository_url>/<image_name>
      

      Copy

      例

      docker push myorg-myacct.registry.snowflakecomputing.com/tutorial_db/data_schema/tutorial_repository/my_echo_service_image:latest
      

      Copy

3: サービスを作成する¶

このセクションでは、サービスを作成し、サービスと通信するためのサービス関数も作成します。

サービスを作成するには、以下が必要です。

• コンピューティングプール。Snowflakeは指定したコンピューティングプールでサービスを実行します。共通セットアップの一環としてコンピューティングプールを作成しました。

• サービス仕様。この仕様は、サービスの構成と実行に必要な情報をSnowflakeに提供します。詳細については、 Snowpark Container Services: サービスの操作 をご参照ください。このチュートリアルでは、 CREATE SERVICE コマンドで仕様をインラインで提供します。また、チュートリアル2に示すように、仕様をSnowflakeステージのファイルに保存し、 CREATE SERVICE コマンドでファイル情報を提供することもできます。

サービス関数は、サービスと通信するために利用できる方法の1つです。サービス関数は、サービスエンドポイントに関連付けるユーザー定義関数（UDF）です。サービス関数が実行されると、サービスエンドポイントにリクエストを送り、応答を受け取ります。

1. コンピューティングプールの準備ができ、サービスを作成するのに適切なコンテキストにいることを確認します。

   1. 以前は 共通セットアップ ステップでコンテキストを設定しました。このステップの SQL ステートメントのコンテキストが正しいことを確認するために、以下を実行します。

   USE ROLE test_role;
   USE DATABASE tutorial_db;
   USE SCHEMA data_schema;
   USE WAREHOUSE tutorial_warehouse;
   

   Copy

   1. 共通セットアップ で作成したコンピューティングプールが準備できていることを確認するために、 DESCRIBE COMPUTE POOL を実行し、 state が ACTIVE または IDLE であることを確認します。 state が STARTING の場合は、 state が ACTIVE か IDLE に変わるまで待つ必要があります。

   DESCRIBE COMPUTE POOL tutorial_compute_pool;
   

   Copy

2. サービスを作成するには、 test_role を使用して以下のコマンドを実行します。

   CREATE SERVICE echo_service
     IN COMPUTE POOL tutorial_compute_pool
     FROM SPECIFICATION $$
       spec:
         containers:
         - name: echo
           image: /tutorial_db/data_schema/tutorial_repository/my_echo_service_image:latest
           env:
             SERVER_PORT: 8000
             CHARACTER_NAME: Bob
           readinessProbe:
             port: 8000
             path: /healthcheck
         endpoints:
         - name: echoendpoint
           port: 8000
           public: true
         $$
      MIN_INSTANCES=1
      MAX_INSTANCES=1;
   

   Copy

   注釈

   その名前のサービスがすでに存在する場合は、 DROP SERVICE コマンドを使用して以前に作成したサービスを削除してから、このサービスを作成します。

3. 以下の SQL コマンドを実行して、先ほど作成したサービスの詳細情報を取得します。詳細については、 Snowpark Container Services: サービスの操作 をご参照ください。

   • アカウント内のサービスをリストするには、 SHOW SERVICES コマンドを実行します。

     SHOW SERVICES;
     

     Copy

   • サービスのステータスを取得するには、システム関数 SYSTEM$GET_SERVICE_STATUS を呼び出します。

     SELECT SYSTEM$GET_SERVICE_STATUS('echo_service');
     

     Copy

   • サービスに関する情報を取得するには、 DESCRIBE SERVICE コマンドを実行します。

     DESCRIBE SERVICE echo_service;
     

     Copy

4. サービス関数を作成するには、以下のコマンドを実行します。

   CREATE FUNCTION my_echo_udf (InputText varchar)
     RETURNS varchar
     SERVICE=echo_service
     ENDPOINT=echoendpoint
     AS '/echo';
   

   Copy

   次の点に注意してください。

   • SERVICE プロパティは、 UDF と echo_service サービスを関連付けます。

   • ENDPOINT プロパティは、サービス内の UDF と echoendpoint エンドポイントを関連付けます。

   • AS 「/echo」は、エコーサーバーへの HTTP パスを指定します。このパスは、サービスコード echo_service.py にあります。

4: サービスを使用する¶

まず、このセクションの SQL ステートメントのコンテキストを設定し、以下を実行します。

USE ROLE test_role;
USE DATABASE tutorial_db;
USE SCHEMA data_schema;
USE WAREHOUSE tutorial_warehouse;

これで、Echoサービスと通信できるようになりました。

1. サービス関数の使用: クエリ内でサービス関数を呼び出すことができます。例のサービス関数（my_echo_udf）は、単一の文字列または文字列のリストを入力として受け取ることができます。

   例1.1: 単一の文字列を渡す

   • my_echo_udf サービス関数を呼び出すには、次の SELECT ステートメントを実行し、1つの入力文字列（'hello'）を渡します。

     SELECT my_echo_udf('hello!');
     

     Copy

     Snowflakeは、 POST リクエストをサービスエンドポイント（echoendpoint）に送信します。リクエストを受信すると、サービスは応答に入力文字列をエコーします。

     +--------------------------+
     | **MY_ECHO_UDF('HELLO!')**|
     |------------------------- |
     | Bob said hello!          |
     +--------------------------+
     

   例1.2: 文字列のリストを渡す

   サービス関数に文字列のリストを渡すと、Snowflakeはこれらの入力文字列を一括して、一連の POST リクエストをサービスに送信します。サービスがすべての文字列を処理した後、Snowflakeは結果を結合して返します。

   以下の例では、テーブル列を入力としてサービス関数に渡します。

   1. 複数の文字列からなるテーブルを作成します。

      CREATE TABLE messages (message_text VARCHAR)
        AS (SELECT * FROM (VALUES ('Thank you'), ('Hello'), ('Hello World')));
      

      Copy

   2. テーブルが作成されたことを確認します。

      SELECT * FROM messages;
      

      Copy

   3. サービス関数を呼び出すには、次の SELECT ステートメントを実行し、入力としてテーブル行を渡します。

      SELECT my_echo_udf(message_text) FROM messages;
      

      Copy

      出力：

      +---------------------------+
      | MY_ECHO_UDF(MESSAGE_TEXT) |
      |---------------------------|
      | Bob said Thank you        |
      | Bob said Hello            |
      | Bob said Hello World      |
      +---------------------------+
      

2. ウェブブラウザーの使用: サービスはエンドポイントを公開します（CREATE SERVICE コマンドで提供されるインライン仕様を参照）。したがって、サービスがインターネットに公開しているウェブ UI にログインし、ウェブブラウザーからサービスにリクエストを送信できます。

   1. サービスが公開しているパブリックエンドポイントの URL を見つけます。

      SHOW ENDPOINTS IN SERVICE echo_service;
      

      Copy

      応答の ingress_url 列は URL を提供します。

      例

      p6bye-myorg-myacct.snowflakecomputing.app
      

   2. /ui をエンドポイント URL に追加し、ウェブブラウザーに貼り付けます。これにより、サービスは ui() 関数を実行します（echo_service.py を参照）。

      エンドポイント URL に初めてアクセスするときは、Snowflakeにログインするようにリクエストされることに注意してください。このテストでは、サービスの作成に使用したユーザーと同じユーザーを使用し、そのユーザーが必要な権限を持っていることを確認します。

      

   3. Input ボックスに文字列「Hello」を入力し、 Return を押します。

      

   注釈

   イングレスエンドポイントにはプログラムでアクセスできます。サンプルコードについては、 公開エンドポイントアクセスおよび認証 をご参照ください。Snowflakeがリクエストをサービスコード内の ui() 関数にルーティングできるように、コード内のエンドポイント URL に /ui を追加する必要があることに注意してください。

5: クリーンアップする¶

チュートリアル2 または チュートリアル3 に進む予定がない場合は、作成した請求対象リソースを削除する必要があります。詳細については、 チュートリアル3 のステップ5をご参照ください。

6: サービスコードの確認¶

このセクションでは、以下のトピックを取り上げます。

• チュートリアル1コードの検証: Echoサービスを実装しているコードファイルを確認します。

• サービス関数に対する理解: このセクションでは、このチュートリアルのサービス関数がどのようにサービスとリンクしているかを説明します。

• ローカルでのイメージの構築とテスト。このセクションでは、Snowflakeアカウントのリポジトリにアップロードする前に、ローカルでDockerイメージをテストする方法について説明します。

from flask import Flask
from flask import request
from flask import make_response
from flask import render_template
import logging
import os
import sys

SERVICE_HOST = os.getenv('SERVER_HOST', '0.0.0.0')
SERVER_PORT = os.getenv('SERVER_PORT', 8080)
CHARACTER_NAME = os.getenv('CHARACTER_NAME', 'I')


def get_logger(logger_name):
  logger = logging.getLogger(logger_name)
  logger.setLevel(logging.DEBUG)
  handler = logging.StreamHandler(sys.stdout)
  handler.setLevel(logging.DEBUG)
  handler.setFormatter(
    logging.Formatter(
      '%(name)s [%(asctime)s] [%(levelname)s] %(message)s'))
  logger.addHandler(handler)
  return logger


logger = get_logger('echo-service')

app = Flask(__name__)


@app.get("/healthcheck")
def readiness_probe():
  return "I'm ready!"


@app.post("/echo")
def echo():
  '''
  Main handler for input data sent by Snowflake.
  '''
  message = request.json
  logger.debug(f'Received request: {message}')

  if message is None or not message['data']:
    logger.info('Received empty message')
    return {}

  # input format:
  #   {"data": [
  #     [row_index, column_1_value, column_2_value, ...],
  #     ...
  #   ]}
  input_rows = message['data']
  logger.info(f'Received {len(input_rows)} rows')

  # output format:
  #   {"data": [
  #     [row_index, column_1_value, column_2_value, ...}],
  #     ...
  #   ]}
  output_rows = [[row[0], get_echo_response(row[1])] for row in input_rows]
  logger.info(f'Produced {len(output_rows)} rows')

  response = make_response({"data": output_rows})
  response.headers['Content-type'] = 'application/json'
  logger.debug(f'Sending response: {response.json}')
  return response


@app.route("/ui", methods=["GET", "POST"])
def ui():
  '''
  Main handler for providing a web UI.
  '''
  if request.method == "POST":
    # getting input in HTML form
    input_text = request.form.get("input")
    # display input and output
    return render_template("basic_ui.html",
      echo_input=input_text,
      echo_reponse=get_echo_response(input_text))
  return render_template("basic_ui.html")


def get_echo_response(input):
  return f'{CHARACTER_NAME} said {input}'

if __name__ == '__main__':
  app.run(host=SERVICE_HOST, port=SERVER_PORT)

ARG BASE_IMAGE=python:3.10-slim-buster
FROM $BASE_IMAGE
COPY echo_service.py ./
COPY templates/ ./templates/
RUN pip install --upgrade pip && \\
pip install flask
CMD ["python", "echo_service.py"]

<!DOCTYPE html>
<html lang="en">
  <head>
    <title>Welcome to echo service!</title>
  </head>
  <body>
    <h1>Welcome to echo service!</h1>
    <form action="{{ url_for("ui") }}" method="post">
      <label for="input">Input:<label><br>
      <input type="text" id="input" name="input"><br>
    </form>
    <h2>Input:</h2>
    {{ echo_input }}
    <h2>Output:</h2>
    {{ echo_reponse }}
  </body>
</html>

spec:
  containers:
  - name: echo
    image: /tutorial_db/data_schema/tutorial_repository/my_echo_service_image:latest
    env:
      SERVER_PORT: 8000
      CHARACTER_NAME: Bob
    readinessProbe:
      port: 8000
      path: /healthcheck
  endpoints:
  - name: echoendpoint
    port: 8000
    public: true

CREATE FUNCTION my_echo_udf (InputText VARCHAR)
  RETURNS VARCHAR
  SERVICE=echo_service
  ENDPOINT=echoendpoint
  AS '/echo';

次の内容¶

これでジョブを実行する チュートリアル2 をテストできます。