External network access and private connectivity on AWS¶
This topic provides configuration details to set up outbound private connectivity to an AWS external service by way of external network access. The primary differences between the outbound public connectivity and outbound private connectivity configurations are that, with private connectivity, you must do the following operations:
Create a private connectivity endpoint. This step requires the ACCOUNTADMIN role.
Create the network rule so the
TYPE
property is set toPRIVATE_HOST_PORT
.
Outbound private connectivity costs¶
You pay for each private connectivity endpoint along with total data processed. For pricing of these items, see the Snowflake Service Consumption Table.
You can explore the cost of these items by filtering on the following service types when querying billing views in the ACCOUNT_USAGE and ORGANIZATION_USAGE schemas:
OUTBOUND_PRIVATELINK_ENDPOINT
OUTBOUND_PRIVATELINK_DATA_PROCESSED
For example, you can query the USAGE_IN_CURRENCY_DAILY view and filter on these service types.
Set up private connectivity to an external Amazon S3 service¶
Call the SYSTEM$PROVISION_PRIVATELINK_ENDPOINT system function to specify Snowflake is connecting to an AWS S3 service, and the hostname to use when connecting to the service:
USE ROLE ACCOUNTADMIN; SELECT SYSTEM$PROVISION_PRIVATELINK_ENDPOINT( 'com.amazonaws.us-west-2.s3', '*.s3.us-west-2.amazonaws.com' );
Note
The asterisk in
*.s3.us-west-2.amazonaws.com
specifies that you can use the endpoint to access multiple S3 buckets.Execute the following SQL statement to create a network rule that allows Snowflake to send requests to an external destination, being sure to set the
TYPE
property toPRIVATE_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');
Execute the following SQL statement to create a security integration for external API authentication:
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';
Execute the following SQL statement to get the
STORAGE_AWS_IAM_USER_ARN
andSTORAGE_AWS_EXTERNAL_ID
values for the IAM user:DESC SECURITY INTEGRATION aws_s3_security_integration;
Using the
STORAGE_AWS_IAM_USER_ARN
andSTORAGE_AWS_EXTERNAL_ID
values, follow Step 5 in Option 1: Configuring a Snowflake storage integration to access Amazon S3 to grant the IAM user access to the Amazon S3 service.Execute the following SQL statement to create a token to use for authentication with the AWS S3 service:
CREATE OR REPLACE SECRET aws_s3_access_token TYPE = CLOUD_PROVIDER_TOKEN API_AUTHENTICATION = aws_s3_security_integration;
Execute the following SQL statement to create an external access integration that uses the network rule and token created in previous steps:
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';
Execute one of the following SQL statements to create a function that can use the external access integration and the token that were previously created:
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' $$;
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; } } $$;
Execute one of the following SQL statements to run the function you created:
SELECT aws_s3_python_function();
SELECT aws_s3_java_function();
Set up private connectivity to an external Amazon Bedrock service¶
Call the SYSTEM$PROVISION_PRIVATELINK_ENDPOINT system function to specify that Snowflake is connecting to the AWS S3 and Amazon Bedrock services, and the hostnames to use when connecting to the services:
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' );
Execute the following SQL statement to create a network rule that allows Snowflake to send requests to an external destination, being sure to set the
TYPE
property toPRIVATE_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');
Execute the following SQL statement to create a security integration for external API authentication:
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';
Execute the following SQL statement to get the
STORAGE_AWS_IAM_USER_ARN
andSTORAGE_AWS_EXTERNAL_ID
values for the IAM user:DESC SECURITY INTEGRATION bedrock_security_integration;
Using the
STORAGE_AWS_IAM_USER_ARN
andSTORAGE_AWS_EXTERNAL_ID
values, follow Step 5 in Option 1: Configuring a Snowflake storage integration to access Amazon S3 to grant the IAM user access to the Amazon Bedrock service.Execute the following SQL statement to create a token to use for authentication with the AWS Bedrock service:
CREATE OR REPLACE SECRET aws_bedrock_access_token TYPE = CLOUD_PROVIDER_TOKEN API_AUTHENTICATION = bedrock_security_integration;
Execute the following SQL statement to create an external access integration that uses the network rule and token created in previous steps:
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 ;
Execute the following SQL statement to create a function that can use the external access integration and the token that were previously created:
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 $$;
Execute the following SQL statement to run the function you created:
SELECT bedrock_private_connectivity_tests();