教程 3：服务到服务的通信¶

简介¶

在本教程中，您会创建 Snowpark Container Services 作业服务，该服务与您在 教程 1 中创建的 Echo 服务进行通信。作业服务运行时，会向 Echo 服务 URL（您在服务规范中提供）发送 POST 请求，并在请求正文中包含“Hello”字符串。Echo 服务返回响应正文中带有“Bob said Hello”字符串的响应。您可以访问作业服务容器日志以验证通信是否成功。

本教程分为两个部分：

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

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

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

  3. 暂存规范文件，该文件为 Snowflake 提供容器配置信息。 除了用于启动容器的镜像名称外，该规范还将环境变量 (SERVICE_URL) 设置为 Echo 服务 URL。应用程序代码读取此环境变量以向 Echo 服务发送请求。

  4. 执行作业服务。使用 EXECUTE JOB SERVICE 命令，您可以通过提供规范文件和 Snowflake 可以运行容器的计算池来执行作业服务。最后，访问作业服务容器中的日志，以验证作业服务和服务之间的通信是否成功。

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

先决条件¶

完成 教程 1 并验证 Echo 服务是否正在运行。

SELECT SYSTEM$GET_SERVICE_STATUS('echo_service', 10);

1：下载服务代码¶

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

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

2. 解压内容，其中包含每个教程的一个目录。Tutorial-3 目录包含以下文件：

   • service_to_service.py

   • Dockerfile

   • service_to_service_spec.yaml

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，请使用 service_to_service:latest。

   示例

   docker build --rm --platform linux/amd64 -t myorg-myacct.registry.snowflakecomputing.cn/tutorial_db/data_schema/tutorial_repository/service_to_service: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/service_to_service:latest
      

      Copy

3：暂存规范文件¶

• 要将您的服务规范文件 (service_to_service_spec.yaml) 上传到暂存区，请使用以下选项之一：

  • Snowsight Web 界面。有关说明，请参阅 为本地文件选择内部暂存区。

  • ** SnowSQL CLI。 ** 执行以下 PUT 命令：

    PUT file://<absolute-path-to-spec.yaml> @tutorial_stage
      AUTO_COMPRESS=FALSE
      OVERWRITE=TRUE;
    

    Copy

  此命令设置 OVERWRITE=TRUE，以便您可以在需要时再次上传文件（例如，如果您修复了规范文件中的错误）。如果 PUT 命令成功执行，则会打印出有关上传文件的信息。

4：执行作业服务¶

现在，您可以测试已创建的 Snowflake 作业服务。执行任务服务时，Snowflake 会将容器中的代码输出到标准输出或标准错误的任何内容作为日志收集。您可以使用 SYSTEM$GET_SERVICE_LOGS 系统功能访问日志。有关更多信息，请参阅 Snowpark Container Services：服务的其他注意事项。

1. 要开始作业服务，请运行 EXECUTE JOB SERVICE 命令：

   EXECUTE JOB SERVICE
     IN COMPUTE POOL tutorial_compute_pool
     NAME=tutorial_db.data_schema.tutorial3_job_service
     FROM @tutorial_stage
     SPEC='service_to_service_spec.yaml';
   

   Copy

   请注意以下事项：

   • FROM 和 SPEC 提供暂存区名称和服务规范文件的名称。

   • COMPUTE_POOL 提供 Snowflake 在其中执行作业服务的计算资源。

   Snowflake 运行规范文件中标识的容器。容器读取 SERVICE_URL 环境变量值 (http://echo-service:8000/echo)，并在 /echo HTTP 路径的端口 8000 处向 Echo 服务发送请求。

   Snowflake 开始作业服务并返回以下输出：

   +------------------------------------------------------------------------+
   | status                                                                 |
   |------------------------------------------------------------------------|
   | Job TUTORIAL3_JOB_SERVICE completed successfully with status: DONE     |
   +------------------------------------------------------------------------+
   

   请注意，响应包含作业服务名称。

2. （可选）作业服务完成后，您可以获得有关已执行的作业服务的更多信息。这对于调试作业服务故障十分有用。要获取作业服务状态，请调用 SYSTEM$GET_SERVICE_STATUS 函数：

   CALL SYSTEM$GET_SERVICE_STATUS('TUTORIAL3_JOB_SERVICE');
   

   Copy

   示例输出：

   [
     {
       "status":"DONE",
       "message":"Container finished",
       "containerName":"main",
       "instanceId":"0",
       "serviceName":"TUTORIAL3_JOB_SERVICE",
       "image":"myorg-myacct.registry.snowflakecomputing.cn/tutorial_db/data_schema/tutorial_repository/service_to_service:latest",
       "restartCount":0,
       "startTime":"2023-01-01T00:00:00Z"
     }
   ]
   

3. 要读取作业服务日志，请调用 SYSTEM$GET_SERVICE_LOGS：

   CALL SYSTEM$GET_SERVICE_LOGS('tutorial_3_job_service', 0, 'main');
   

   Copy

   main 是您从中检索日志的容器的名称。您在服务规范文件中为容器设置此容器名称。

   示例日志：

   +--------------------------------------------------------------------------------------------------------------------------+
   | SYSTEM$GET_JOB_LOGS                                                                                                      |
   |--------------------------------------------------------------------------------------------------------------------------|
   | service-to-service [2023-04-29 21:52:09,208] [INFO] Calling http://echo-service:8000/echo with input Hello               |
   | service-to-service [2023-04-29 21:52:09,212] [INFO] Received response from http://echo-service:8000/echo: Bob said Hello |
   +--------------------------------------------------------------------------------------------------------------------------+
   

5：清理¶

Snowflake 会对您的账户中处于活动状态的计算池节点收费。（请参阅 使用计算池）。为防止不必要的费用，请先停止当前在计算池上运行的所有服务。然后，暂停计算池（如果您打算稍后再次使用），或者将其删除。

1. 停止计算池上的所有服务和作业服务。

   ALTER COMPUTE POOL tutorial_compute_pool STOP ALL;
   

   Copy

2. 删除计算池。

   DROP COMPUTE POOL tutorial_compute_pool;
   

   Copy

您也可以清理镜像注册表（移除所有镜像）和内部暂存区（移除规范）。

DROP IMAGE REPOSITORY tutorial_repository;
DROP STAGE tutorial_stage;

6：审查作业服务代码¶

本部分包括以下主题：

• 检查提供的文件：审查用于实施作业服务的各种代码文件。

• 在本地构建和测试镜像。了解如何在将 Docker 镜像上传到 Snowflake 账户中的仓库之前进行本地测试。

import json
import logging
import os
import requests
import sys

SERVICE_URL = os.getenv('SERVICE_URL', 'http://localhost:8080/echo')
ECHO_TEXT = 'Hello'

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('service-to-service')

def call_service(service_url, echo_input):
  logger.info(f'Calling {service_url} with input {echo_input}')

  row_to_send = {"data": [[0, echo_input]]}
  response = requests.post(url=service_url,
                           data=json.dumps(row_to_send),
                           headers={"Content-Type": "application/json"})

  message = response.json()
  if message is None or not message["data"]:
    logger.error('Received empty response from service ' + service_url)

  response_row = message["data"][0]
  if len(response_row) != 2:
    logger.error('Unexpected response format: ' + response_row)

  echo_reponse = response_row[1]
  logger.info(f'Received response from {service_url}: ' + echo_reponse)

if __name__ == '__main__':
  call_service(SERVICE_URL, ECHO_TEXT)

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

spec:
container:
   - name: main
      image: /tutorial_db/data_schema/tutorial_repository/service_to_service:latest
      env:
      SERVICE_URL: "http://echo-service:8000/echo"

下一步是什么？¶

教程 4：创建已挂载块存储卷的服务