AWS 上的外部网络访问和专用连接¶

本主题提供配置详细信息，通过外部网络访问设置对 AWS 外部服务的出站专用连接。出站公共连接和出站专用连接配置之间的主要区别在于，使用专用连接时，您必须执行以下操作：

创建专用连接端点。此步骤需要 ACCOUNTADMIN 角色。
创建网络规则，以便将 TYPE 属性设置为 PRIVATE_HOST_PORT。

出站专用连接成本¶

您为每个专用连接端点以及处理的总数据量付费。有关这些项目的定价，请参阅 Snowflake 服务使用量表。

在 ACCOUNT_USAGE 和 ORGANIZATION_USAGE 架构中查询计费视图时，可以通过筛选以下服务类型来查看这些项目的成本：

OUTBOUND_PRIVATELINK_ENDPOINT
OUTBOUND_PRIVATELINK_DATA_PROCESSED

例如，您可以查询 USAGE_IN_CURRENCY_DAILY 视图并筛选这些服务类型。

设置到外部 Amazon S3 服务的专用连接：¶

调用 SYSTEM$PROVISION_PRIVATELINK_ENDPOINT 系统函数以指定 Snowflake 正连接到 AWS S3 服务，以及连接到服务时使用的主机名：
```
USE ROLE ACCOUNTADMIN;
SELECT SYSTEM$PROVISION_PRIVATELINK_ENDPOINT(
  'com.amazonaws.us-west-2.s3',
  '*.s3.us-west-2.amazonaws.com'
);
```
Copy
备注

*.s3.us-west-2.amazonaws.com 中的星号指定您可以使用端点访问多个 S3 桶。

执行以下 SQL 语句，以创建允许 Snowflake 向外部目标发送请求的网络规则，确保将 TYPE 属性设置为 PRIVATE_HOST_PORT：

CREATE OR REPLACE NETWORK RULE aws_s3_network_rule
  MODE = EGRESS
  TYPE = PRIVATE_HOST_PORT
  VALUE_LIST = ('external-access-iam-bucket.s3.us-west-2.amazonaws.com');

Copy

执行以下 SQL 语句，以创建针对外部 API 身份验证的安全集成：

CREATE OR REPLACE SECURITY INTEGRATION aws_s3_security_integration
  TYPE = API_AUTHENTICATION
  AUTH_TYPE = AWS_IAM
  ENABLED = TRUE
  AWS_ROLE_ARN = 'arn:aws:iam::736112632310:role/external-access-iam-bucket';

Copy

执行以下 SQL 语句，以获取针对 IAM 用户的 STORAGE_AWS_IAM_USER_ARN 和 STORAGE_AWS_EXTERNAL_ID 值：
```
DESC SECURITY INTEGRATION aws_s3_security_integration;
```
Copy
使用 STORAGE_AWS_IAM_USER_ARN 和 STORAGE_AWS_EXTERNAL_ID 值，按照选项 1：配置 Snowflake 存储集成以访问 Amazon S3 中 第 5 步 授予对 Amazon S3 服务的 IAM 用户访问权限。

执行以下 SQL 语句以创建令牌，用于 AWS S3 服务的身份验证：

CREATE OR REPLACE SECRET aws_s3_access_token
  TYPE = CLOUD_PROVIDER_TOKEN
  API_AUTHENTICATION = aws_s3_security_integration;

Copy

执行以下 SQL 语句以创建外部访问集成，使用前面步骤中创建的网络规则和令牌：

CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION aws_s3_external_access_integration
  ALLOWED_NETWORK_RULES = (aws_s3_network_rule)
  ALLOWED_AUTHENTICATION_SECRETS = (aws_s3_access_token)
  ENABLED = TRUE
  COMMENT = 'Testing S3 connectivity';

Copy

执行以下 SQL 语句之一，以创建可以使用外部访问集成和以前创建的令牌的函数：

CREATE OR REPLACE FUNCTION aws_s3_python_function()
  RETURNS VARCHAR
  LANGUAGE PYTHON
  EXTERNAL_ACCESS_INTEGRATIONS = (aws_s3_external_access_integration)
  RUNTIME_VERSION = '3.8'
  SECRETS = ('cred' = aws_s3_access_token)
  PACKAGES = ('boto3')
  HANDLER = 'main_handler'
AS
$$
  import boto3
  import _snowflake
  from botocore.config import Config

  def main_handler():
      # Get the previously created token as an object
      cloud_provider_object = _snowflake.get_cloud_provider_token('cred')

      # Configure boto3 connection settings
      config = Config(
          retries=dict(total_max_attempts=9),
          connect_timeout=30,
          read_timeout=30,
          max_pool_connections=50
      )

      # Connect to S3 using boto3
      s3 = boto3.client(
          's3',
          region_name='us-west-2',
          aws_access_key_id=cloud_provider_object.access_key_id,
          aws_secret_access_key=cloud_provider_object.secret_access_key,
          aws_session_token=cloud_provider_object.token,
          config=config
      )

      # Use the s3 object upload/download resources
      # ...

      return 'Successfully connected to AWS S3'
$$;

Copy

CREATE OR REPLACE FUNCTION aws_s3_java_function()
  RETURNS STRING
  LANGUAGE JAVA
  EXTERNAL_ACCESS_INTEGRATIONS = (aws_s3_external_access_integration)
  SECRETS = ('cred' = aws_s3_access_token)
  HANDLER = 'AWSTokenProvider.handle'
AS
$$
  import com.snowflake.snowpark_java.types.CloudProviderToken;
  import com.snowflake.snowpark_java.types.SnowflakeSecrets;

  public class AWSTokenProvider {
      public static String handle() {
          // Get the previously created token as an object
          SnowflakeSecrets sfSecret = SnowflakeSecrets.newInstance();
          CloudProviderToken cloudProviderToken = sfSecret.getCloudProviderToken("cred");

          // Create variables for the AWS session credentials
          String accessKeyId = cloudProviderToken.getAccessKeyId();
          String secretAccessKey = cloudProviderToken.getSecretAccessKey();
          String token = cloudProviderToken.getToken();

          // Use the token to create an S3 client
          // ...

          return "Successfully connected to AWS S3 with the following access token: " + token;
      }
  }
$$;

Copy

执行以下 SQL 语句之一，以运行您创建的函数：
SELECT aws_s3_python_function();
Copy
SELECT aws_s3_java_function();
Copy

设置与外部 Amazon Bedrock 服务的专用连接¶

调用 SYSTEM$PROVISION_PRIVATELINK_ENDPOINT 系统函数，指定 Snowflake 正连接到 AWS S3 和 Amazon Bedrock 服务，以及连接到服务时使用的主机名：

USE ROLE ACCOUNTADMIN;
SELECT SYSTEM$PROVISION_PRIVATELINK_ENDPOINT(
  'com.amazonaws.us-west-2.s3',
  '*.s3.us-west-2.amazonaws.com'
);

SELECT SYSTEM$PROVISION_PRIVATELINK_ENDPOINT(
  'com.amazonaws.us-west-2.bedrock-runtime',
  'bedrock-runtime.us-west-2.amazonaws.com'
);

Copy

执行以下 SQL 语句，以创建允许 Snowflake 向外部目标发送请求的网络规则，确保将 TYPE 属性设置为 PRIVATE_HOST_PORT：

CREATE OR REPLACE NETWORK RULE bedrock_network_rule
  MODE = EGRESS
  TYPE = PRIVATE_HOST_PORT
  VALUE_LIST = ('bedrock-runtime.us-west-2.amazonaws.com');

Copy

执行以下 SQL 语句，以创建针对外部 API 身份验证的安全集成：

CREATE OR REPLACE SECURITY INTEGRATION bedrock_security_integration
  TYPE = API_AUTHENTICATION
  AUTH_TYPE = AWS_IAM
  ENABLED = TRUE
  AWS_ROLE_ARN = 'arn:aws:iam::736112632310:role/external-access-iam-bucket';

Copy

执行以下 SQL 语句，以获取针对 IAM 用户的 STORAGE_AWS_IAM_USER_ARN 和 STORAGE_AWS_EXTERNAL_ID 值：
```
DESC  SECURITY INTEGRATION bedrock_security_integration;
```
Copy
使用 STORAGE_AWS_IAM_USER_ARN 和 STORAGE_AWS_EXTERNAL_ID 值，按照选项 1：配置 Snowflake 存储集成以访问 Amazon S3 中 第 5 步 授予对 Amazon Bedrock 服务的 IAM 用户访问权限。

执行以下 SQL 语句以创建令牌，用于 AWS Bedrock 服务的身份验证：

CREATE OR REPLACE SECRET aws_bedrock_access_token
  TYPE = CLOUD_PROVIDER_TOKEN
  API_AUTHENTICATION = bedrock_security_integration;

Copy

执行以下 SQL 语句以创建外部访问集成，使用前面步骤中创建的网络规则和令牌：

CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION bedrock_external_access_integration
  ALLOWED_NETWORK_RULES = (bedrock_network_rule)
  ALLOWED_AUTHENTICATION_SECRETS=(aws_bedrock_access_token)
  ENABLED=true ;

Copy

执行以下 SQL 语句，以创建可以使用外部访问集成和以前创建的令牌的函数：

CREATE OR REPLACE FUNCTION bedrock_private_connectivity_tests(
  id INT,
  instructions VARCHAR,
  user_context VARCHAR,
  model_id VARCHAR
)
  RETURNS VARCHAR
  LANGUAGE PYTHON
  EXTERNAL_ACCESS_INTEGRATIONS = (bedrock_external_access_integration)
  RUNTIME_VERSION = '3.8'
  SECRETS = ('cred' = aws_bedrock_access_token)
  PACKAGES = ('boto3')
  HANDLER = 'bedrock_py'
AS
$$
  import boto3
  import json
  import _snowflake
  def bedrock_py(id, instructions, user_context, model_id):
      # Get the previously created token as an object
      cloud_provider_object = _snowflake.get_cloud_provider_token('cred')
      cloud_provider_dictionary = {
          "ACCESS_KEY_ID": cloud_provider_object.access_key_id,
          "SECRET_ACCESS_KEY": cloud_provider_object.secret_access_key,
          "TOKEN": cloud_provider_object.token
      }
      # Assign AWS credentials and choose a region
      boto3_session_args = {
          'aws_access_key_id': cloud_provider_dictionary["ACCESS_KEY_ID"],
          'aws_secret_access_key': cloud_provider_dictionary["SECRET_ACCESS_KEY"],
          'aws_session_token': cloud_provider_dictionary["TOKEN"],
          'region_name': 'us-west-2'
      }
      session = boto3.Session(**boto3_session_args)
      client = session.client('bedrock-runtime')
      # Prepare the request body for the specified model
      def prepare_request_body(model_id, instructions, user_context):
          default_max_tokens = 512
          default_temperature = 0.7
          default_top_p = 1.0
          if model_id == 'amazon.titan-text-express-v1':
              body = {
                  "inputText": f"<SYSTEM>Follow these:{instructions}<END_SYSTEM>\n<USER_CONTEXT>Use this user context in your response:{user_context}<END_USER_CONTEXT>",
                  "textGenerationConfig": {
                      "maxTokenCount": default_max_tokens,
                      "stopSequences": [],
                      "temperature": default_temperature,
                      "topP": default_top_p
                  }
              }
          elif model_id == 'ai21.j2-ultra-v1':
              body = {
                  "prompt": f"<SYSTEM>Follow these:{instructions}<END_SYSTEM>\n<USER_CONTEXT>Use this user context in your response:{user_context}<END_USER_CONTEXT>",
                  "temperature": default_temperature,
                  "topP": default_top_p,
                  "maxTokens": default_max_tokens
              }
          elif model_id == 'anthropic.claude-3-sonnet-20240229-v1:0':
              body = {
                  "max_tokens": default_max_tokens,
                  "messages": [{"role": "user", "content": f"<SYSTEM>Follow these:{instructions}<END_SYSTEM>\n<USER_CONTEXT>Use this user context in your response:{user_context}<END_USER_CONTEXT>"}],
                  "anthropic_version": "bedrock-2023-05-31"
              }
          else:
              raise ValueError("Unsupported model ID")
          return json.dumps(body)
      # Call Bedrock to get a completion
      body = prepare_request_body(model_id, instructions, user_context)
      response = client.invoke_model(modelId=model_id, body=body)
      response_body = json.loads(response.get('body').read())
      # Parse the API response based on the model
      def get_completion_from_response(response_body, model_id):
          if model_id == 'amazon.titan-text-express-v1':
              output_text = response_body.get('results')[0].get('outputText')
          elif model_id == 'ai21.j2-ultra-v1':
              output_text = response_body.get('completions')[0].get('data').get('text')
          elif model_id == 'anthropic.claude-3-sonnet-20240229-v1:0':
              output_text = response_body.get('content')[0].get('text')
          else:
              raise ValueError("Unsupported model ID")
          return output_text
      # Get the generated text from Bedrock
      output_text = get_completion_from_response(response_body, model_id)
      return output_text
  $$;

Copy

执行以下 SQL 语句，以运行您创建的函数：
```
SELECT bedrock_private_connectivity_tests();
```
Copy