教程 1：创建 Snowpark Container Services 服务¶

简介¶

完成 通用设置 后，您就可以创建服务了。在本教程中，您将创建一项服务（名为 echo_service），该服务仅回显您以输入形式提供的文本。例如，如果输入字符串为“Hello World”，则服务将返回“I said， Hello World”。

本教程分为两个部分：

第 1 部分：创建并测试服务。 您将下载为本教程提供的代码，并按照分步说明进行操作：

1. 下载为本教程提供的服务代码。

2. 为 Snowpark Container Services 构建 Docker 镜像，并将该镜像上传到账户的仓库中。

3. 提供服务规范文件和用于运行服务的计算池，以创建服务。

4. 创建服务函数，与服务进行通信。

5. 使用服务。向服务发送回显请求并验证响应。

第 2 部分：了解服务。本部分概述服务代码，并重点介绍不同组件的协作方式。

1：下载服务代码¶

提供用于创建 Echo 服务的代码（Python 应用程序）。

1. 下载：download: SnowparkContainerServices -Tutorials.zip </samples/spcs/SnowparkContainerServices-Tutorials.zip>。

2. 解压内容，其中包含每个教程的一个目录。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.cn/tutorial_db/data_schema/tutorial_repository
     

   • 仓库 URL 中的主机名是注册表主机名。示例如下：

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

构建镜像并将其上传到仓库

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.cn/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.cn/tutorial_db/data_schema/tutorial_repository/my_echo_service_image:latest
      

      Copy

3：创建服务¶

在本部分中，您将创建一项服务，并创建服务函数以与服务进行通信。

要创建服务，您需要满足以下条件：

• 计算池。Snowflake 在指定的计算池中运行您的服务。您在通用设置中创建了一个计算池。

• 服务规范。此规范为 Snowflake 提供了配置和运行服务所需的信息。有关更多信息，请参阅 Snowpark Container Services：使用服务。在本教程中，您将在 CREATE SERVICE 命令以内联方式提供规范。您还可以将规范保存到 Snowflake 暂存区的文件中，并在 CREATE SERVICE 命令中提供文件信息，如教程 2 中所示。

服务函数是用于与服务进行通信的方法之一。服务函数是用户定义的函数 (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' 指定 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 语句，传递一个输入字符串 ('hello')：

     SELECT my_echo_udf('hello!');
     

     Copy

     Snowflake 向服务端点 (echoendpoint) 发送 POST 请求。收到请求后，服务会在响应中回显输入字符串。

     +--------------------------+
     | **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. 使用 Web 浏览器： 服务会公开端点（请参阅 CREATE SERVICE 命令中提供的内联规范）。因此，您可以登录服务向互联网公开的 Web UI，然后从 Web 浏览器向该服务发送请求。

   1. 找到服务公开的公众端点的 URL：

      SHOW ENDPOINTS IN SERVICE echo_service;
      

      Copy

      响应中的 ingress_url 列提供 URL。

      示例

      p6bye-myorg-myacct.snowflakecomputing.app
      

   2. 将 /ui 追加到端点 URL，并将其粘贴到 Web 浏览器中。这样会导致服务执行 ui() 函数（请参阅 echo_service.py）。

      请注意，首次访问端点 URL 时，系统将要求您登录 Snowflake。对于此测试，请使用用于创建服务的同一用户，以确保用户具有必要的权限。

      

   3. 在 输入 框中输入字符串“Hello”，然后按 Return。

      

   备注

   您可以通过编程方式访问入口端点。有关示例代码，请参阅 公共端点访问和身份验证。请注意，您需要将 /ui 追加到代码中的端点 URL，以便 Snowflake 可以将请求路由到服务代码中的 ui() 函数。

5：清理¶

如果您不打算继续学习 教程 2 或 教程 3，则应移除您创建的计费资源。有关更多信息，请参阅 教程 3 中的第 5 步。

6：查看服务代码¶

本部分包括以下主题：

• 检查教程 1 代码：查看用于实施 Echo 服务的代码文件。

• 了解服务函数：本部分介绍本教程中的服务函数如何与服务链接。

• 在本地构建和测试镜像。本部分介绍如何本地测试 Docker 镜像，然后将其上传到 Snowflake 账户的仓库。

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。