使用适用于 Python 的软件开发工具包的 Amazon Redshift 示例 (Boto3) - AWS SDK 代码示例

文档 AWS SDK 示例 GitHub 存储库中还有更多 S AWS DK 示例

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

使用适用于 Python 的软件开发工具包的 Amazon Redshift 示例 (Boto3)

以下代码示例向您展示了如何在 Amazon Redshift 中 AWS SDK for Python (Boto3) 使用来执行操作和实现常见场景。

基础知识是向您展示如何在服务中执行基本操作的代码示例。

操作是大型程序的代码摘录,必须在上下文中运行。您可以通过操作了解如何调用单个服务函数,还可以通过函数相关场景的上下文查看操作。

每个示例都包含一个指向完整源代码的链接,您可以从中找到有关如何在上下文中设置和运行代码的说明。

开始使用

以下代码示例展示了如何开始使用 Amazon Redshift。

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

import boto3 def hello_redshift(redshift_client): """ Use the AWS SDK for Python (Boto3) to create an Amazon Redshift client and list the clusters in your account. This list might be empty if you haven't created any clusters. This example uses the default settings specified in your shared credentials and config files. :param redshift_client: A Boto3 Redshift Client object. """ print("Hello, Redshift! Let's list your clusters:") paginator = redshift_client.get_paginator("describe_clusters") clusters = [] for page in paginator.paginate(): clusters.extend(page["Clusters"]) print(f"{len(clusters)} cluster(s) were found.") for cluster in clusters: print(f" {cluster['ClusterIdentifier']}") if __name__ == "__main__": hello_redshift(boto3.client("redshift"))
  • 有关 API 的详细信息,请参阅适用DescribeClustersPython 的AWS SDK (Boto3) API 参考

基本功能

以下代码示例展示了如何使用软件开发工具包学习 Amazon Redshift 的 AWS 核心操作。

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

class RedshiftScenario: """Runs an interactive scenario that shows how to get started with Redshift.""" def __init__(self, redshift_wrapper, redshift_data_wrapper): self.redshift_wrapper = redshift_wrapper self.redshift_data_wrapper = redshift_data_wrapper def redhift_scenario(self, json_file_path): database_name = "dev" print(DASHES) print("Welcome to the Amazon Redshift SDK Getting Started example.") print( """ This Python program demonstrates how to interact with Amazon Redshift using the AWS SDK for Python (Boto3). Amazon Redshift is a fully managed, petabyte-scale data warehouse service hosted in the cloud. The program's primary functionalities include cluster creation, verification of cluster readiness, listing databases, table creation, populating data within the table, and executing SQL statements. It also demonstrates querying data from the Movies table. Upon completion, all AWS resources are cleaned up. """ ) if not os.path.isfile(json_file_path): logging.error(f"The file {json_file_path} does not exist.") return print("Let's get started...") user_name = q.ask("Please enter your user name (default is awsuser):") user_name = user_name if user_name else "awsuser" print(DASHES) user_password = q.ask( "Please enter your user password (default is AwsUser1000):" ) user_password = user_password if user_password else "AwsUser1000" print(DASHES) print( """A Redshift cluster refers to the collection of computing resources and storage that work together to process and analyze large volumes of data.""" ) cluster_id = q.ask( "Enter a cluster identifier value (default is redshift-cluster-movies): " ) cluster_id = cluster_id if cluster_id else "redshift-cluster-movies" self.redshift_wrapper.create_cluster( cluster_id, "ra3.4xlarge", user_name, user_password, True, 2 ) print(DASHES) print(f"Wait until {cluster_id} is available. This may take a few minutes...") q.ask("Press Enter to continue...") self.wait_cluster_available(cluster_id) print(DASHES) print( f""" When you created {cluster_id}, the dev database is created by default and used in this scenario. To create a custom database, you need to have a CREATEDB privilege. For more information, see the documentation here: https://docs.aws.amazon.com/redshift/latest/dg/r_CREATE_DATABASE.html. """ ) q.ask("Press Enter to continue...") print(DASHES) print(DASHES) print(f"List databases in {cluster_id}") q.ask("Press Enter to continue...") databases = self.redshift_data_wrapper.list_databases( cluster_id, database_name, user_name ) print(f"The cluster contains {len(databases)} database(s).") for database in databases: print(f" Database: {database}") print(DASHES) print(DASHES) print("Now you will create a table named Movies.") q.ask("Press Enter to continue...") self.create_table(cluster_id, database_name, user_name) print(DASHES) print("Populate the Movies table using the Movies.json file.") print( "Specify the number of records you would like to add to the Movies Table." ) print("Please enter a value between 50 and 200.") while True: try: num_records = int(q.ask("Enter a value: ", q.is_int)) if 50 <= num_records <= 200: break else: print("Invalid input. Please enter a value between 50 and 200.") except ValueError: print("Invalid input. Please enter a value between 50 and 200.") self.populate_table( cluster_id, database_name, user_name, json_file_path, num_records ) print(DASHES) print("Query the Movies table by year. Enter a value between 2012-2014.") while True: movie_year = int(q.ask("Enter a year: ", q.is_int)) if 2012 <= movie_year <= 2014: break else: print("Invalid input. Please enter a valid year between 2012 and 2014.") # Function to query database sql_id = self.query_movies_by_year( database_name, user_name, movie_year, cluster_id ) print(f"The identifier of the statement is {sql_id}") print("Checking statement status...") self.wait_statement_finished(sql_id) result = self.redshift_data_wrapper.get_statement_result(sql_id) self.display_movies(result) print(DASHES) print(DASHES) print("Now you will modify the Redshift cluster.") q.ask("Press Enter to continue...") preferred_maintenance_window = "wed:07:30-wed:08:00" self.redshift_wrapper.modify_cluster(cluster_id, preferred_maintenance_window) print(DASHES) print(DASHES) delete = q.ask("Do you want to delete the cluster? (y/n) ", q.is_yesno) if delete: print(f"You selected to delete {cluster_id}") q.ask("Press Enter to continue...") self.redshift_wrapper.delete_cluster(cluster_id) else: print(f"Cluster {cluster_id}cluster_id was not deleted") print(DASHES) print("This concludes the Amazon Redshift SDK Getting Started scenario.") print(DASHES) def create_table(self, cluster_id, database, username): self.redshift_data_wrapper.execute_statement( cluster_identifier=cluster_id, database_name=database, user_name=username, sql="CREATE TABLE Movies (statement_id INT PRIMARY KEY, title VARCHAR(100), year INT)", ) print("Table created: Movies") def populate_table(self, cluster_id, database, username, file_name, number): with open(file_name) as f: data = json.load(f) i = 0 for record in data: if i == number: break statement_id = i title = record["title"] year = record["year"] i = i + 1 parameters = [ {"name": "statement_id", "value": str(statement_id)}, {"name": "title", "value": title}, {"name": "year", "value": str(year)}, ] self.redshift_data_wrapper.execute_statement( cluster_identifier=cluster_id, database_name=database, user_name=username, sql="INSERT INTO Movies VALUES(:statement_id, :title, :year)", parameter_list=parameters, ) print(f"{i} records inserted into Movies table") def wait_cluster_available(self, cluster_id): """ Waits for a cluster to be available. :param cluster_id: The cluster identifier. Note: The cluster_available waiter can also be used. It is not used in this case to allow an elapsed time message. """ cluster_ready = False start_time = time.time() while not cluster_ready: time.sleep(30) cluster = self.redshift_wrapper.describe_clusters(cluster_id) status = cluster[0]["ClusterStatus"] if status == "available": cluster_ready = True elif status != "creating": raise Exception( f"Cluster {cluster_id} creation failed with status {status}." ) elapsed_seconds = int(round(time.time() - start_time)) minutes = int(elapsed_seconds // 60) seconds = int(elapsed_seconds % 60) print(f"Elapsed Time: {minutes}:{seconds:02d} - status {status}...") if minutes > 30: raise Exception( f"Cluster {cluster_id} is not available after 30 minutes." ) def query_movies_by_year(self, database, username, year, cluster_id): sql = "SELECT * FROM Movies WHERE year = :year" params = [{"name": "year", "value": str(year)}] response = self.redshift_data_wrapper.execute_statement( cluster_identifier=cluster_id, database_name=database, user_name=username, sql=sql, parameter_list=params, ) return response["Id"] @staticmethod def display_movies(response): metadata = response["ColumnMetadata"] records = response["Records"] title_column_index = None for i in range(len(metadata)): if metadata[i]["name"] == "title": title_column_index = i break if title_column_index is None: print("No title column found.") return print(f"Found {len(records)} movie(s).") for record in records: print(f" {record[title_column_index]['stringValue']}") def wait_statement_finished(self, sql_id): while True: time.sleep(1) response = self.redshift_data_wrapper.describe_statement(sql_id) status = response["Status"] print(f"Statement status is {status}.") if status == "FAILED": print(f"The query failed because {response['Error']}. Ending program") raise Exception("The Query Failed. Ending program") elif status == "FINISHED": break

用于显示场景实施的主函数。

def main(): redshift_client = boto3.client("redshift") redshift_data_client = boto3.client("redshift-data") redshift_wrapper = RedshiftWrapper(redshift_client) redshift_data_wrapper = RedshiftDataWrapper(redshift_data_client) redshift_scenario = RedshiftScenario(redshift_wrapper, redshift_data_wrapper) redshift_scenario.redhift_scenario( f"{os.path.dirname(__file__)}/../../../resources/sample_files/movies.json" )

场景中使用的包装器函数。

def create_cluster( self, cluster_identifier, node_type, master_username, master_user_password, publicly_accessible, number_of_nodes, ): """ Creates a cluster. :param cluster_identifier: The name of the cluster. :param node_type: The type of node in the cluster. :param master_username: The master username. :param master_user_password: The master user password. :param publicly_accessible: Whether the cluster is publicly accessible. :param number_of_nodes: The number of nodes in the cluster. :return: The cluster. """ try: cluster = self.client.create_cluster( ClusterIdentifier=cluster_identifier, NodeType=node_type, MasterUsername=master_username, MasterUserPassword=master_user_password, PubliclyAccessible=publicly_accessible, NumberOfNodes=number_of_nodes, ) return cluster except ClientError as err: logging.error( "Couldn't create a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def describe_clusters(self, cluster_identifier): """ Describes a cluster. :param cluster_identifier: The cluster identifier. :return: A list of clusters. """ try: kwargs = {} if cluster_identifier: kwargs["ClusterIdentifier"] = cluster_identifier paginator = self.client.get_paginator("describe_clusters") clusters = [] for page in paginator.paginate(**kwargs): clusters.extend(page["Clusters"]) return clusters except ClientError as err: logging.error( "Couldn't describe a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def execute_statement( self, cluster_identifier, database_name, user_name, sql, parameter_list=None ): """ Executes a SQL statement. :param cluster_identifier: The cluster identifier. :param database_name: The database name. :param user_name: The user's name. :param sql: The SQL statement. :param parameter_list: The optional SQL statement parameters. :return: The SQL statement result. """ try: kwargs = { "ClusterIdentifier": cluster_identifier, "Database": database_name, "DbUser": user_name, "Sql": sql, } if parameter_list: kwargs["Parameters"] = parameter_list response = self.client.execute_statement(**kwargs) return response except ClientError as err: logging.error( "Couldn't execute statement. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def describe_statement(self, statement_id): """ Describes a SQL statement. :param statement_id: The SQL statement identifier. :return: The SQL statement result. """ try: response = self.client.describe_statement(Id=statement_id) return response except ClientError as err: logging.error( "Couldn't describe statement. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def get_statement_result(self, statement_id): """ Gets the result of a SQL statement. :param statement_id: The SQL statement identifier. :return: The SQL statement result. """ try: result = { "Records": [], } paginator = self.client.get_paginator("get_statement_result") for page in paginator.paginate(Id=statement_id): if "ColumnMetadata" not in result: result["ColumnMetadata"] = page["ColumnMetadata"] result["Records"].extend(page["Records"]) return result except ClientError as err: logging.error( "Couldn't get statement result. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def modify_cluster(self, cluster_identifier, preferred_maintenance_window): """ Modifies a cluster. :param cluster_identifier: The cluster identifier. :param preferred_maintenance_window: The preferred maintenance window. """ try: self.client.modify_cluster( ClusterIdentifier=cluster_identifier, PreferredMaintenanceWindow=preferred_maintenance_window, ) except ClientError as err: logging.error( "Couldn't modify a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def list_databases(self, cluster_identifier, database_name, database_user): """ Lists databases in a cluster. :param cluster_identifier: The cluster identifier. :param database_name: The database name. :param database_user: The database user. :return: The list of databases. """ try: paginator = self.client.get_paginator("list_databases") databases = [] for page in paginator.paginate( ClusterIdentifier=cluster_identifier, Database=database_name, DbUser=database_user, ): databases.extend(page["Databases"]) return databases except ClientError as err: logging.error( "Couldn't list databases. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def delete_cluster(self, cluster_identifier): """ Deletes a cluster. :param cluster_identifier: The cluster identifier. """ try: self.client.delete_cluster( ClusterIdentifier=cluster_identifier, SkipFinalClusterSnapshot=True ) except ClientError as err: logging.error( "Couldn't delete a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

操作

以下代码示例演示如何使用 CreateCluster

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

class RedshiftWrapper: """ Encapsulates Amazon Redshift cluster operations. """ def __init__(self, redshift_client): """ :param redshift_client: A Boto3 Redshift client. """ self.client = redshift_client def create_cluster( self, cluster_identifier, node_type, master_username, master_user_password, publicly_accessible, number_of_nodes, ): """ Creates a cluster. :param cluster_identifier: The name of the cluster. :param node_type: The type of node in the cluster. :param master_username: The master username. :param master_user_password: The master user password. :param publicly_accessible: Whether the cluster is publicly accessible. :param number_of_nodes: The number of nodes in the cluster. :return: The cluster. """ try: cluster = self.client.create_cluster( ClusterIdentifier=cluster_identifier, NodeType=node_type, MasterUsername=master_username, MasterUserPassword=master_user_password, PubliclyAccessible=publicly_accessible, NumberOfNodes=number_of_nodes, ) return cluster except ClientError as err: logging.error( "Couldn't create a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

以下代码实例化对象。 RedshiftWrapper

client = boto3.client("redshift") redhift_wrapper = RedshiftWrapper(client)
  • 有关 API 的详细信息,请参阅适用CreateClusterPython 的AWS SDK (Boto3) API 参考

以下代码示例演示如何使用 DeleteCluster

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

class RedshiftWrapper: """ Encapsulates Amazon Redshift cluster operations. """ def __init__(self, redshift_client): """ :param redshift_client: A Boto3 Redshift client. """ self.client = redshift_client def delete_cluster(self, cluster_identifier): """ Deletes a cluster. :param cluster_identifier: The cluster identifier. """ try: self.client.delete_cluster( ClusterIdentifier=cluster_identifier, SkipFinalClusterSnapshot=True ) except ClientError as err: logging.error( "Couldn't delete a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

以下代码实例化对象。 RedshiftWrapper

client = boto3.client("redshift") redhift_wrapper = RedshiftWrapper(client)
  • 有关 API 的详细信息,请参阅适用DeleteClusterPython 的AWS SDK (Boto3) API 参考

以下代码示例演示如何使用 DescribeClusters

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

class RedshiftWrapper: """ Encapsulates Amazon Redshift cluster operations. """ def __init__(self, redshift_client): """ :param redshift_client: A Boto3 Redshift client. """ self.client = redshift_client def describe_clusters(self, cluster_identifier): """ Describes a cluster. :param cluster_identifier: The cluster identifier. :return: A list of clusters. """ try: kwargs = {} if cluster_identifier: kwargs["ClusterIdentifier"] = cluster_identifier paginator = self.client.get_paginator("describe_clusters") clusters = [] for page in paginator.paginate(**kwargs): clusters.extend(page["Clusters"]) return clusters except ClientError as err: logging.error( "Couldn't describe a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

以下代码实例化对象。 RedshiftWrapper

client = boto3.client("redshift") redhift_wrapper = RedshiftWrapper(client)
  • 有关 API 的详细信息,请参阅适用DescribeClustersPython 的AWS SDK (Boto3) API 参考

以下代码示例演示如何使用 DescribeStatement

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

class RedshiftDataWrapper: """Encapsulates Amazon Redshift data.""" def __init__(self, client): """ :param client: A Boto3 RedshiftDataWrapper client. """ self.client = client def describe_statement(self, statement_id): """ Describes a SQL statement. :param statement_id: The SQL statement identifier. :return: The SQL statement result. """ try: response = self.client.describe_statement(Id=statement_id) return response except ClientError as err: logging.error( "Couldn't describe statement. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

以下代码实例化对象。 RedshiftDataWrapper

client = boto3.client("redshift-data") redshift_data_wrapper = RedshiftDataWrapper(client)
  • 有关 API 的详细信息,请参阅适用DescribeStatementPython 的AWS SDK (Boto3) API 参考

以下代码示例演示如何使用 GetStatementResult

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

class RedshiftDataWrapper: """Encapsulates Amazon Redshift data.""" def __init__(self, client): """ :param client: A Boto3 RedshiftDataWrapper client. """ self.client = client def get_statement_result(self, statement_id): """ Gets the result of a SQL statement. :param statement_id: The SQL statement identifier. :return: The SQL statement result. """ try: result = { "Records": [], } paginator = self.client.get_paginator("get_statement_result") for page in paginator.paginate(Id=statement_id): if "ColumnMetadata" not in result: result["ColumnMetadata"] = page["ColumnMetadata"] result["Records"].extend(page["Records"]) return result except ClientError as err: logging.error( "Couldn't get statement result. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

以下代码实例化对象。 RedshiftDataWrapper

client = boto3.client("redshift-data") redshift_data_wrapper = RedshiftDataWrapper(client)
  • 有关 API 的详细信息,请参阅适用GetStatementResultPython 的AWS SDK (Boto3) API 参考

以下代码示例演示如何使用 ModifyCluster

适用于 Python 的 SDK(Boto3)
注意

还有更多相关信息 GitHub。查找完整示例,学习如何在 AWS 代码示例存储库中进行设置和运行。

class RedshiftWrapper: """ Encapsulates Amazon Redshift cluster operations. """ def __init__(self, redshift_client): """ :param redshift_client: A Boto3 Redshift client. """ self.client = redshift_client def modify_cluster(self, cluster_identifier, preferred_maintenance_window): """ Modifies a cluster. :param cluster_identifier: The cluster identifier. :param preferred_maintenance_window: The preferred maintenance window. """ try: self.client.modify_cluster( ClusterIdentifier=cluster_identifier, PreferredMaintenanceWindow=preferred_maintenance_window, ) except ClientError as err: logging.error( "Couldn't modify a cluster. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise

以下代码实例化对象。 RedshiftWrapper

client = boto3.client("redshift") redhift_wrapper = RedshiftWrapper(client)
  • 有关 API 的详细信息,请参阅适用ModifyClusterPython 的AWS SDK (Boto3) API 参考