AWS 에서의 외부 네트워크 액세스 및 비공개 연결¶

외부 네트워크 액세스 를 통해 AWS 외부 서비스에 대한 아웃바운드 비공개 연결을 위해 Snowflake를 구성할 수 있습니다.

공용 연결과 달리 비공개 연결에서는 다음 작업을 수행해야 합니다.

비공개 연결 엔드포인트를 생성합니다. 이 단계에는 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'
);
```
참고

*.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');
```

다음 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';

다음 SQL 문을 실행하여 IAM 사용자에 대한 STORAGE_AWS_IAM_USER_ARN 및 STORAGE_AWS_EXTERNAL_ID 값을 가져옵니다.
```
DESC SECURITY INTEGRATION aws_s3_security_integration;
```
STORAGE_AWS_IAM_USER_ARN 및 STORAGE_AWS_EXTERNAL_ID 값을 사용하여 Option 1: Configure a Snowflake storage integration to access Amazon S3 에서 5단계 를 따라 IAM 사용자에게 Amazon S3 서비스에 대한 액세스 권한을 부여합니다.

다음 SQL 문을 실행하여 AWS S3 서비스 인증에 사용할 토큰을 생성합니다.

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

다음 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';

다음 SQL 문 중 하나를 실행하여 외부 액세스 통합과 이전에 생성한 토큰을 사용할 수 있는 함수를 만듭니다.

CREATE OR REPLACE FUNCTION aws_s3_python_function()
  RETURNS VARCHAR
  LANGUAGE PYTHON
  EXTERNAL_ACCESS_INTEGRATIONS = (aws_s3_external_access_integration)
  RUNTIME_VERSION = '3.12'
  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'
$$;

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;
      }
  }
$$;

다음 SQL 문 중 하나를 실행하여 생성한 함수를 실행합니다.
SELECT aws_s3_python_function();
SELECT aws_s3_java_function();

외부 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'
);

다음 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');
```

다음 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';

다음 SQL 문을 실행하여 IAM 사용자에 대한 STORAGE_AWS_IAM_USER_ARN 및 STORAGE_AWS_EXTERNAL_ID 값을 가져옵니다.
```
DESC  SECURITY INTEGRATION bedrock_security_integration;
```
STORAGE_AWS_IAM_USER_ARN 및 STORAGE_AWS_EXTERNAL_ID 값을 사용하여 Option 1: Configure a Snowflake storage integration to access Amazon S3 에서 5단계 를 따라 IAM 사용자에게 Amazon Bedrock 서비스에 대한 액세스 권한을 부여합니다.

다음 SQL 문을 실행하여 AWS Bedrock 서비스 인증에 사용할 토큰을 생성합니다.

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

다음 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 ;

다음 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
  $$;

다음 SQL 문을 실행하여 생성한 함수를 실행합니다.

SELECT bedrock_private_connectivity_tests(1, 'Summarize the main benefits of attending this university.', 'University of Waterloo', 'amazon.titan-text-express-v1');