Snowpark Container Services：SQL 执行¶

凭据配置选项¶

Snowflake 建议应用程序容器在执行 SQL 时，使用 Snowflake 提供的凭据进行身份验证。虽然可以通过外部访问集成 (EAI) 使用其他凭据，但通过 EAI 连接会将服务视为在 Snowflake 外部运行，并经由互联网连接到 Snowflake。

您可以从服务容器通过以下三种选项连接到 Snowflake：

• 使用 Snowflake 提供的服务用户凭据： Snowflake 为每个服务提供凭据，称为服务凭据。服务使用这些凭据以“服务用户”身份连接到 Snowflake。

• 使用 Snowflake 提供的调用方凭据（调用方权限）： 当您使用调用方权限配置服务时，Snowflake 还会为服务提供凭据，以便以“调用用户”身份连接到 Snowflake。

• 使用其他凭据： 在这种情况下，您使用外部访问集成 (EAI)，允许您的服务通过有效的身份验证凭据连接到 Snowflake 的互联网端点。此选项需要管理员创建 EAI，然后将该集成的 USAGE 权限授予服务所有者角色。

  备注

  如果您使用外部访问集成访问 Snowflake，您可能会通过互联网发送潜在的敏感信息。

有关使用各种 Snowflake 驱动程序连接到 Snowflake 的代码示例，请参阅 Snowflake 连接示例 (https://github.com/Snowflake-Labs/sf-samples/tree/main/samples/spcs/sf-connection)。

SNOWFLAKE_ACCOUNT = os.getenv('SNOWFLAKE_ACCOUNT')
SNOWFLAKE_HOST = os.getenv('SNOWFLAKE_HOST')

def get_login_token():
  with open('/snowflake/session/token', 'r') as f:
    return f.read()

conn = snowflake.connector.connect(
  host = os.getenv('SNOWFLAKE_HOST'),
  account = os.getenv('SNOWFLAKE_ACCOUNT'),
  token = get_login_token(),
  authenticator = 'oauth'
)

ALTER USER my_user SET DEFAULT_SECONDARY_ROLES = ( 'ALL' );

-- Permissions to resolve the table's name.
GRANT CALLER USAGE ON DATABASE <db_name> TO ROLE <service_owner_role>;
GRANT CALLER USAGE ON SCHEMA <schema_name> TO ROLE <service_owner_role>;
-- Permissions to use a warehouse
GRANT CALLER USAGE ON WAREHOUSE <warehouse_name> TO ROLE <service_owner_role>;
-- Permissions to query the table.
GRANT CALLER SELECT ON TABLE T1 TO ROLE <service_owner_role>;

conn = snowflake.connector.connect(
  account = '<acct-name>',
  user = '<user-name>',
  password = '<password>'
)

配置用于执行 SQL 的数据库和架构上下文¶

除了提供凭据外，Snowflake 还提供创建该服务时所在的数据库和架构上下文。容器代码可以使用此信息，在与服务相同的数据库和架构上下文中执行 SQL。

本节介绍两个概念：

• Snowflake 用于确定要在其中创建服务的数据库和架构的逻辑。

• Snowflake 将此信息传达给容器，从而使容器代码能够在同一数据库和架构上下文中执行 SQL 的方法。

Snowflake 使用服务名称来确定要在其中创建服务的数据库和架构：

• 示例 1：在以下 CREATE SERVICE 和 EXECUTE JOB SERVICE 命令中，服务名称不会显式指定数据库和架构名称。Snowflake 在当前数据库和架构中创建服务和作业服务。

  -- Create a service.
  CREATE SERVICE test_service IN COMPUTE POOL ...
  
  -- Execute a job service.
  EXECUTE JOB SERVICE
    IN COMPUTE POOL tutorial_compute_pool
    NAME = example_job_service ...
  

  Copy

• 示例 2：在以下 CREATE SERVICE 和 EXECUTE JOB SERVICE 命令中，服务名称包括数据库和架构名称。Snowflake 在指定的数据库 (test_db) 和架构 (test_schema) 中创建服务和作业服务，无论当前架构如何。

  -- Create a service.
  CREATE SERVICE test_db.test_schema.test_service IN COMPUTE POOL ...
  
  -- Execute a job service.
  EXECUTE JOB SERVICE
    IN COMPUTE POOL tutorial_compute_pool
    NAME = test_db.test_schema.example_job_service ...
  

  Copy

当 Snowflake 启动服务时，它会使用以下环境变量向正在运行的容器提供数据库和架构信息：

• SNOWFLAKE_DATABASE

• SNOWFLAKE_SCHEMA

容器代码可以使用连接代码中的环境变量来确定要使用的数据库和架构，如以下示例所示：

conn = snowflake.connector.connect(
  host = os.getenv('SNOWFLAKE_HOST'),
  account = os.getenv('SNOWFLAKE_ACCOUNT'),
  token = get_login_token(),
  authenticator = 'oauth',
  database = os.getenv('SNOWFLAKE_DATABASE'),
  schema = os.getenv('SNOWFLAKE_SCHEMA')
)

示例

在 教程 2 中，您可以创建与 Snowflake 连接并执行 SQL 语句的 Snowflake 作业服务。以下步骤总结了教程代码如何使用环境变量：

1. 在通用设置中（请参阅 通用设置 部分），您可以创建资源，包括数据库和架构。您还可以为会话设置当前数据库和架构：

   USE DATABASE tutorial_db;
   ...
   USE SCHEMA data_schema;
   

   Copy

2. 创建作业服务（通过运行 EXECUTE JOB SERVICE）后，Snowflake 启动容器，并将容器中的以下环境变量设置为会话的当前数据库和架构：

   • 将 SNOWFLAKE_DATABASE 设置为“TUTORIAL_DB”

   • 将 SNOWFLAKE_SCHEMA 设置为“DATA_SCHEMA”

3. 作业代码（请参阅教程 2 中的 main.py）读取以下环境变量：

   SNOWFLAKE_DATABASE = os.getenv('SNOWFLAKE_DATABASE')
   SNOWFLAKE_SCHEMA = os.getenv('SNOWFLAKE_SCHEMA')
   

   Copy

4. 作业代码将数据库和架构设置为执行 SQL 语句（main.py 中的 run_job() 函数）的上下文:

   {
      "account": SNOWFLAKE_ACCOUNT,
      "host": SNOWFLAKE_HOST,
      "authenticator": "oauth",
      "token": get_login_token(),
      "warehouse": SNOWFLAKE_WAREHOUSE,
      "database": SNOWFLAKE_DATABASE,
      "schema": SNOWFLAKE_SCHEMA
   }
   ...
   

   Copy

   备注

   SNOWFLAKE_ACCOUNT、SNOWFLAKE_HOST、SNOWFLAKE_DATABASE、SNOWFLAKE_SCHEMA 是 Snowflake 为应用程序容器生成的环境变量，但 SNOWFLAKE_WAREHOUSE 不是（教程 2 应用程序代码创建了此变量，因为 Snowflake 不会将仓库名称传递给容器）。

为容器指定仓库¶

如果服务连接到 Snowflake，以在 Snowflake 仓库中执行查询，则您可以使用以下选项来指定仓库：

• 在应用程序代码中指定仓库。 在启动 Snowflake 会话以在代码中运行查询时，在连接配置中指定仓库。有关示例，请参阅 教程 2。

• 在创建服务时指定默认仓库。 在 CREATE SERVICE 或 EXECUTE JOB SERVICE 命令中指定可选的 QUERY_WAREHOUSE 参数以提供默认仓库。如果您的应用程序代码未在连接配置中提供仓库，Snowflake 将使用默认仓库。如果应用程序代码未在连接配置中提供仓库，则 Snowflake 将使用默认仓库。使用 ALTER SERVICE 命令更改默认仓库。

  备注

  使用 QUERY_WAREHOUSE 参数指定的仓库仅作为 服务用户 的默认仓库。当服务代表另一个用户连接到 Snowflake 时（在 调用方权限场景 中），Snowflake 会使用该用户的默认仓库。

如果同时通过这两种方法指定了仓库，则将使用应用程序代码中指定的仓库。

访问服务用户查询历史记录¶

您可以通过筛选 QUERY_HISTORY 视图 或 QUERY_HISTORY 函数（其中 user_type 为 SNOWFLAKE_SERVICE）来查找由您的服务以服务用户身份执行的查询。

示例 1： 提取一项服务运行的查询。

SELECT *
FROM snowflake.account_usage.query_history
WHERE user_type = 'SNOWFLAKE_SERVICE'
AND user_name = '<service_name>'
AND user_database_name = '<service_db_name>'
AND user_schema_name = '<service_schema_name>'
order by start_time;

在 WHERE 子句中：

• user_name = '<service_name>'：您将服务名称指定为用户名，是因为服务以 服务用户 的身份执行查询，且服务用户的名称与服务名称相同。

• user_type = 'SNOWFLAKE_SERVICE' 和 user_name = '<service_name>'：这样会限制查询结果，即仅检索一项服务执行的查询。

• user_database_name 和 user_schema_name：对于服务用户而言，这些名称是服务的数据库和架构。

调用 QUERY_HISTORY 函数可以得到相同的结果。

SELECT *
FROM TABLE(<service_db_name>.information_schema.query_history())
WHERE user_database_name = '<service_db_name>'
AND user_schema_name = '<service_schema_name>'
AND user_type = 'SNOWFLAKE_SERVICE'
AND user_name = '<service_name>'
order by start_time;

在 WHERE 子句中：

• user_type = 'SNOWFLAKE_SERVICE' 和 user_name = '<service_name>' 会限制查询结果，仅检索一项服务执行的查询。

• 对于服务用户而言，user_database_name 和 user_schema_name 名称是服务的数据库和架构。

示例 2： 获取多项服务运行的查询和相应的服务信息。

SELECT query_history.*, services.*
FROM snowflake.account_usage.query_history
JOIN snowflake.account_usage.services
ON query_history.user_name = services.service_name
AND query_history.user_schema_id = services.service_schema_id
AND query_history.user_type = 'SNOWFLAKE_SERVICE'

该查询联接 QUERY_HISTORY 和 SERVICES 视图，以检索与查询和执行查询的服务相关的信息。请注意以下事项：

• 对于由服务运行的查询，query_history.user_name 是服务用户的名称，与服务名称相同。

• 该查询使用架构 IDs（而非架构名称）联接视图，以确保您引用相同的架构，因为如果删除并重新创建架构，架构 ID 会发生变化，但名称保持不变。

您可以向查询添加可选的筛选器。例如：

• 筛选 query_history，从而仅检索执行特定查询的服务。

• 筛选 services，从而仅检索特定服务执行的查询。

示例 3： 获取每项服务的服务用户信息。

SELECT services.*, users.*
FROM snowflake.account_usage.users
JOIN snowflake.account_usage.services
ON users.name = services.service_name
AND users.schema_id = services.service_schema_id
AND users.type = 'SNOWFLAKE_SERVICE'

查询在 ACCOUNT_USAGE 架构中联接 SERVICES 和 USERS 视图，以检索服务和服务用户信息。请注意以下事项：

• 当服务运行查询时，它会以服务用户身份运行查询，服务用户名与服务名称相同。因此，您可以指定联接条件：users.name = services.service_name。

• 服务名称仅在一个架构内唯一。因此，该查询指定联接条件 (users.schema_id = services.service_schema_id)，以确保每个服务用户均与其所属的特定服务（而非在不同架构中运行的任何其他同名服务）相匹配。