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

先决条件¶

确保您拥有以下信息：

• 公共端点的入口 URL。 执行 SHOW ENDPOINTS IN SERVICE 命令来获取 URL：

  SHOW ENDPOINTS IN SERVICE echo_service;
  

  Copy

• 您的 Snowflake 账户名称。 有关更多信息，请参阅 常见设置：确认您已准备好继续。

• 您的 Snowflake 账户 URL： 是 <acctname>.snowflakecomputing.cn。

• Snowflake 账户中的用户名。 这是您在 常见设置：创建 Snowflake 对象 中选择的用户。您以该用户身份登录 Snowflake 并测试编程访问。

• 角色名称： 您创建了一个角色 (test_role)，作为常见设置的一部分。用户使用此角色来执行操作。

设置¶

按照步骤，以编程方式与 Echo 服务通信。使用提供的 Python 代码，您可以向 Echo 服务公开的公共端点发送请求。

1. 在命令提示符处，创建目录并导航到该目录。

2. 为用户配置密钥对身份验证。

   1. 生成 密钥对：

      1. 生成私钥。为了简化练习步骤，生成一个未加密的私钥。您也可以使用加密私钥，但需要您输入密码。

         openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
         

         Copy

      2. 通过引用您创建的私钥生成公钥 (rsa_key.pub)。

         openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
         

         Copy

   2. 验证您是否已在目录中生成私钥和公钥。

   3. 将公钥分配给您用来测试编程访问的用户。这使用户可以指定用于身份验证的密钥。

      ALTER USER <user-name> SET RSA_PUBLIC_KEY='MIIBIjANBgkqh...';
      

      Copy

3. 将提供的示例代码保存在 Python 文件中。

   1. 将以下代码保存到 generateJWT.py。

      # To run this on the command line, enter:
      #   python3 generateJWT.py --account=<account_identifier> --user=<username> --private_key_file_path=<path_to_private_key_file>
      
      from cryptography.hazmat.primitives.serialization import load_pem_private_key
      from cryptography.hazmat.primitives.serialization import Encoding
      from cryptography.hazmat.primitives.serialization import PublicFormat
      from cryptography.hazmat.backends import default_backend
      from datetime import timedelta, timezone, datetime
      import argparse
      import base64
      from getpass import getpass
      import hashlib
      import logging
      import sys
      
      # This class relies on the PyJWT module (https://pypi.org/project/PyJWT/).
      import jwt
      
      logger = logging.getLogger(__name__)
      
      try:
          from typing import Text
      except ImportError:
          logger.debug('# Python 3.5.0 and 3.5.1 have incompatible typing modules.', exc_info=True)
          from typing_extensions import Text
      
      ISSUER = "iss"
      EXPIRE_TIME = "exp"
      ISSUE_TIME = "iat"
      SUBJECT = "sub"
      
      # If you generated an encrypted private key, implement this method to return
      # the passphrase for decrypting your private key. As an example, this function
      # prompts the user for the passphrase.
      def get_private_key_passphrase():
          return getpass('Passphrase for private key: ')
      
      class JWTGenerator(object):
          """
          Creates and signs a JWT with the specified private key file, username, and account identifier. The JWTGenerator keeps the
          generated token and only regenerates the token if a specified period of time has passed.
          """
          LIFETIME = timedelta(minutes=59)  # The tokens will have a 59-minute lifetime
          RENEWAL_DELTA = timedelta(minutes=54)  # Tokens will be renewed after 54 minutes
          ALGORITHM = "RS256"  # Tokens will be generated using RSA with SHA256
      
          def __init__(self, account: Text, user: Text, private_key_file_path: Text,
                      lifetime: timedelta = LIFETIME, renewal_delay: timedelta = RENEWAL_DELTA):
              """
              __init__ creates an object that generates JWTs for the specified user, account identifier, and private key.
              :param account: Your Snowflake account identifier. See https://docs.snowflake.com/en/user-guide/admin-account-identifier.html. Note that if you are using the account locator, exclude any region information from the account locator.
              :param user: The Snowflake username.
              :param private_key_file_path: Path to the private key file used for signing the JWTs.
              :param lifetime: The number of minutes (as a timedelta) during which the key will be valid.
              :param renewal_delay: The number of minutes (as a timedelta) from now after which the JWT generator should renew the JWT.
              """
      
              logger.info(
                  """Creating JWTGenerator with arguments
                  account : %s, user : %s, lifetime : %s, renewal_delay : %s""",
                  account, user, lifetime, renewal_delay)
      
              # Construct the fully qualified name of the user in uppercase.
              self.account = self.prepare_account_name_for_jwt(account)
              self.user = user.upper()
              self.qualified_username = self.account + "." + self.user
      
              self.lifetime = lifetime
              self.renewal_delay = renewal_delay
              self.private_key_file_path = private_key_file_path
              self.renew_time = datetime.now(timezone.utc)
              self.token = None
      
              # Load the private key from the specified file.
              with open(self.private_key_file_path, 'rb') as pem_in:
                  pemlines = pem_in.read()
                  try:
                      # Try to access the private key without a passphrase.
                      self.private_key = load_pem_private_key(pemlines, None, default_backend())
                  except TypeError:
                      # If that fails, provide the passphrase returned from get_private_key_passphrase().
                      self.private_key = load_pem_private_key(pemlines, get_private_key_passphrase().encode(), default_backend())
      
          def prepare_account_name_for_jwt(self, raw_account: Text) -> Text:
              """
              Prepare the account identifier for use in the JWT.
              For the JWT, the account identifier must not include the subdomain or any region or cloud provider information.
              :param raw_account: The specified account identifier.
              :return: The account identifier in a form that can be used to generate the JWT.
              """
              account = raw_account
              if not '.global' in account:
                  # Handle the general case.
                  idx = account.find('.')
                  if idx > 0:
                      account = account[0:idx]
              else:
                  # Handle the replication case.
                  idx = account.find('-')
                  if idx > 0:
                      account = account[0:idx]
              # Use uppercase for the account identifier.
              return account.upper()
      
          def get_token(self) -> Text:
              """
              Generates a new JWT. If a JWT has already been generated earlier, return the previously generated token unless the
              specified renewal time has passed.
              :return: the new token
              """
              now = datetime.now(timezone.utc)  # Fetch the current time
      
              # If the token has expired or doesn't exist, regenerate the token.
              if self.token is None or self.renew_time <= now:
                  logger.info("Generating a new token because the present time (%s) is later than the renewal time (%s)",
                              now, self.renew_time)
                  # Calculate the next time we need to renew the token.
                  self.renew_time = now + self.renewal_delay
      
                  # Prepare the fields for the payload.
                  # Generate the public key fingerprint for the issuer in the payload.
                  public_key_fp = self.calculate_public_key_fingerprint(self.private_key)
      
                  # Create our payload
                  payload = {
                      # Set the issuer to the fully qualified username concatenated with the public key fingerprint.
                      ISSUER: self.qualified_username + '.' + public_key_fp,
      
                      # Set the subject to the fully qualified username.
                      SUBJECT: self.qualified_username,
      
                      # Set the issue time to now.
                      ISSUE_TIME: now,
      
                      # Set the expiration time, based on the lifetime specified for this object.
                      EXPIRE_TIME: now + self.lifetime
                  }
      
                  # Regenerate the actual token
                  token = jwt.encode(payload, key=self.private_key, algorithm=JWTGenerator.ALGORITHM)
                  # If you are using a version of PyJWT prior to 2.0, jwt.encode returns a byte string instead of a string.
                  # If the token is a byte string, convert it to a string.
                  if isinstance(token, bytes):
                    token = token.decode('utf-8')
                  self.token = token
                  logger.info("Generated a JWT with the following payload: %s", jwt.decode(self.token, key=self.private_key.public_key(), algorithms=[JWTGenerator.ALGORITHM]))
      
              return self.token
      
          def calculate_public_key_fingerprint(self, private_key: Text) -> Text:
              """
              Given a private key in PEM format, return the public key fingerprint.
              :param private_key: private key string
              :return: public key fingerprint
              """
              # Get the raw bytes of public key.
              public_key_raw = private_key.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
      
              # Get the sha256 hash of the raw bytes.
              sha256hash = hashlib.sha256()
              sha256hash.update(public_key_raw)
      
              # Base64-encode the value and prepend the prefix 'SHA256:'.
              public_key_fp = 'SHA256:' + base64.b64encode(sha256hash.digest()).decode('utf-8')
              logger.info("Public key fingerprint is %s", public_key_fp)
      
              return public_key_fp
      
      def main():
          logging.basicConfig(stream=sys.stdout, level=logging.INFO)
          cli_parser = argparse.ArgumentParser()
          cli_parser.add_argument('--account', required=True, help='The account identifier (e.g. "myorganization-myaccount" for "myorganization-myaccount.snowflakecomputing.cn").')
          cli_parser.add_argument('--user', required=True, help='The user name.')
          cli_parser.add_argument('--private_key_file_path', required=True, help='Path to the private key file used for signing the JWT.')
          cli_parser.add_argument('--lifetime', type=int, default=59, help='The number of minutes that the JWT should be valid for.')
          cli_parser.add_argument('--renewal_delay', type=int, default=54, help='The number of minutes before the JWT generator should produce a new JWT.')
          args = cli_parser.parse_args()
      
          token = JWTGenerator(args.account, args.user, args.private_key_file_path, timedelta(minutes=args.lifetime), timedelta(minutes=args.renewal_delay)).get_token()
          print('JWT:')
          print(token)
      
      if __name__ == "__main__":
          main()
      

      Copy

   2. 将以下代码保存到 access-via-keypair.py。

      from generateJWT import JWTGenerator
      from datetime import timedelta
      import argparse
      import logging
      import sys
      import requests
      logger = logging.getLogger(__name__)
      
      def main():
        args = _parse_args()
        token = _get_token(args)
        snowflake_jwt = token_exchange(token,endpoint=args.endpoint, role=args.role,
                        snowflake_account_url=args.snowflake_account_url,
                        snowflake_account=args.account)
        spcs_url=f'https://{args.endpoint}{args.endpoint_path}'
        connect_to_spcs(snowflake_jwt, spcs_url)
      
      def _get_token(args):
        token = JWTGenerator(args.account, args.user, args.private_key_file_path, timedelta(minutes=args.lifetime),
                  timedelta(minutes=args.renewal_delay)).get_token()
        logger.info("Key Pair JWT: %s" % token)
        return token
      
      def token_exchange(token, role, endpoint, snowflake_account_url, snowflake_account):
        scope_role = f'session:role:{role}' if role is not None else None
        scope = f'{scope_role} {endpoint}' if scope_role is not None else endpoint
        data = {
          'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
          'scope': scope,
          'assertion': token,
        }
        logger.info(data)
        url = f'https://{snowflake_account}.snowflakecomputing.cn/oauth/token'
        if snowflake_account_url:
          url =       f'{snowflake_account_url}/oauth/token'
        logger.info("oauth url: %s" %url)
        response = requests.post(url, data=data)
        logger.info("snowflake jwt : %s" % response.text)
        assert 200 == response.status_code, "unable to get snowflake token"
        return response.text
      
      def connect_to_spcs(token, url):
        # Create a request to the ingress endpoint with authz.
        headers = {'Authorization': f'Snowflake Token="{token}"'}
        response = requests.post(f'{url}', headers=headers)
        logger.info("return code %s" % response.status_code)
        logger.info(response.text)
      
      def _parse_args():
        logging.basicConfig(stream=sys.stdout, level=logging.INFO)
        cli_parser = argparse.ArgumentParser()
        cli_parser.add_argument('--account', required=True,
                    help='The account identifier (for example, "myorganization-myaccount" for '
                      '"myorganization-myaccount.snowflakecomputing.cn").')
        cli_parser.add_argument('--user', required=True, help='The user name.')
        cli_parser.add_argument('--private_key_file_path', required=True,
                    help='Path to the private key file used for signing the JWT.')
        cli_parser.add_argument('--lifetime', type=int, default=59,
                    help='The number of minutes that the JWT should be valid for.')
        cli_parser.add_argument('--renewal_delay', type=int, default=54,
                    help='The number of minutes before the JWT generator should produce a new JWT.')
        cli_parser.add_argument('--role',
                    help='The role we want to use to create and maintain a session for. If a role is not provided, '
                      'use the default role.')
        cli_parser.add_argument('--endpoint', required=True,
                    help='The ingress endpoint of the service')
        cli_parser.add_argument('--endpoint-path', default='/',
                    help='The url path for the ingress endpoint of the service')
        cli_parser.add_argument('--snowflake_account_url', default=None,
                    help='The account url of the account for which we want to log in. Type of '
                      'https://myorganization-myaccount.snowflakecomputing.cn')
        args = cli_parser.parse_args()
        return args
      
      if __name__ == "__main__":
        main()
      

      Copy

以编程方式向服务端点发送请求¶

执行 access-via-keypair.py Python 代码，对 Echo 服务公共端点进行入口调用。

python3 access-via-keypair.py \
  --account <account-identifier> \
  --user <user-name> \
  --role TEST_ROLE \
  --private_key_file_path rsa_key.p8 \
  --endpoint <ingress-hostname> \
  --endpoint-path /ui

Copy

有关 account-identifier 的更多信息，请参阅 账户标识符。

身份验证的工作原理¶

代码首先将提供的密钥对转换为 JWT 令牌。然后发送 JWT 令牌到 Snowflake 以获取 OAuth 令牌。最后，代码使用 OAuth 令牌连接到 Snowflake 并访问公共端点。具体而言，代码执行以下操作：

1. 调用 _get_token(args) 函数，从您提供的密钥对生成 JWT 令牌。函数实施如图所示：

   def _get_token(args):
       token = JWTGenerator(args.account,
                           args.user,
                           args.private_key_file_path,
                           timedelta(minutes=args.lifetime),
                           timedelta(minutes=args.renewal_delay)).get_token()
       logger.info("Key Pair JWT: %s" % token)
       return token
   

   Copy

   JWTGenerator 是一个向您提供的辅助函数类。请注意创建此对象时提供的以下参数：

   • args.account 和 args.user 参数：JWT 令牌有几个字段（请参阅 令牌格式)，iss 是其中一个字段。该字段值包括 Snowflake 账户名称和用户名。因此，您提供这些值作为参数。

   • 这两个 timedelta 参数提供以下信息：

     • lifetime 指定密钥有效的分钟数（60 分钟）。

     • renewal_delay 指定从现在起多少分钟后，JWT 生成器应续期 JWT。

2. 调用 token_exchange() 函数以连接到 Snowflake 并交换 JWT 令牌，以获取 OAuth 令牌。

   scope_role = f'session:role:{role}' if role is not None else None
   scope = f'{scope_role} {endpoint}' if scope_role is not None else endpoint
   
   data = {
       'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
       'scope': scope,
       'assertion': token,
   }
   

   Copy

   上述代码构造了一个 JSON，设置了 OAuth 令牌的范围，以及使用指定角色可以访问的公共端点。然后，此代码会向 Snowflake 发出 POST 请求，传递 JSON 以交换 JWT 令牌，获取 OAuth 令牌（请参阅 令牌交换），如图所示：

   url = f'{snowflake_account_url}/oauth/token'
   response = requests.post(url, data=data)
   assert 200 == response.status_code, "unable to get Snowflake token"
   return response.text
   

   Copy

3. 然后代码调用 connect_to_spcs() 函数以连接到 Echo 服务的公共端点。它提供了端点 URL (https://<ingress-URL>/ui) 和用于身份验证的 OAuth 令牌。

   headers = {'Authorization': f'Snowflake Token="{token}"'}
   response = requests.post(f'{url}', headers=headers)
   

   Copy

   url 是您提供给程序的 spcs_url，token 是 OAuth 令牌。

   本示例中的 Echo 服务提供 HTML 页面（如上一部分所述）。此示例代码在响应中仅打印 HTML。

检查教程 1 代码¶

您在第 1 步中下载的 Zip 文件包含以下文件：

• Dockerfile

• echo_service.py

• templates/basic_ui.html

在创建服务时，您还可以使用服务规范。以下部分介绍这些代码组件如何协同工作以创建服务。

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

了解服务函数¶

服务函数是用于与服务进行通信的方法之一（请参阅 使用服务）。服务函数是用户定义的函数 (UDF)，与服务端点关联。当执行服务函数时，它会向关联的服务端点发送请求并接收响应。

您可以执行具有以下参数的 CREATE FUNCTION 命令，从而创建以下服务函数：

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

请注意以下事项：

• my_echo_udf 函数将字符串作为输入并返回字符串。

• SERVICE 属性标识服务 (echo_service)，ENDPOINT 属性标识用户友好的端点名称 (echoendpoint)。

• AS '/echo' 指定服务的路径。在 echo_service.py 中，@app.post 装饰器将此路径与 echo 函数关联。

此函数连接指定 SERVICE 的特定 ENDPOINT。当您调用此函数时，Snowflake 会向服务容器内的 /echo 路径发送请求。

在本地构建和测试镜像¶

您可以在本地测试 Docker 镜像，然后再将其上传到 Snowflake 账户中的仓库。在本地测试中，容器独立运行（它不是 Snowflake 运行的服务）。

要测试教程 1 Docker 镜像，请执行以下步骤：

1. 要在 Docker CLI 中创建 Docker 镜像，请执行以下命令：

   docker build --rm -t my_service:local .
   

   Copy

2. 要启动代码，请执行以下命令：

   docker run --rm -p 8080:8080 my_service:local
   

   Copy

3. 使用以下方法之一，向服务发送回显请求：

   • 使用 cURL 命令：

     在另一个终端窗口中，使用 cURL，将以下 POST 请求发送到端口 8080：

     curl -X POST http://localhost:8080/echo \
       -H "Content-Type: application/json" \
       -d '{"data":[[0, "Hello friend"], [1, "Hello World"]]}'
     

     Copy

     请注意，请求正文包含两个字符串。cURL 命令将 POST 请求发送到服务正在侦听的端口 8080。数据中的 0 是列表中输入字符串的索引。Echo 服务在响应中回显输入字符串，如下所示：

     {"data":[[0,"I said Hello Friend"],[1,"I said Hello World"]]}
     

   • 使用 Web 浏览器：

     1. 在在同一台计算机上的浏览器中，打开 http://localhost:8080/ui。

        这会将 GET 请求发送到服务正在侦听的端口 8080。服务执行 ui() 函数，该函数呈现 HTML 表单，如下所示：

        

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