使用亚马逊 Redshift 数据 API 与亚马逊 Redshift 无服务器进行交互

Amazon Redshift 是一个快速、可扩展、安全且完全托管的云数据仓库，可使用标准 SQL 和现有 ETL（提取、转换和加载）、商业智能 (BI) 和报告工具轻松且经济高效地分析所有数据。成千上万的客户使用 Amazon Redshift 每天处理艾字节的数据，为 BI、预测分析和实时流分析等分析工作负载提供支持。 Amazon Redshift 无服务器 使您可以方便地运行和扩展分析，而无需预置和管理数据仓库。借助 Redshift Serverless，数据分析师、开发人员和数据科学家现在可以使用 Amazon Redshift，通过向数据仓库加载数据和从数据仓库中查询记录，在几秒钟内从数据中获得见解。

作为数据工程师或应用程序开发人员，对于某些用例，您希望与 Redshift Serverless 数据仓库进行交互，以便使用简单的 API 端点加载或查询数据，而无需管理持续连接。使用 亚马逊 Redshift 数据 API ，您无需配置 JDBC 或 ODBC 即可与 Redshift Serverless 进行交互。这使得使用Redshift Serverless变得更容易、更安全，并开辟了新的用例。

这篇文章解释了如何通过 亚马逊云科技 命令行接口 (亚马逊云科技 CLI) 和 Python 将数据 API 与 Redshift Serverless 一起使用。如果您想将数据 API 用于亚马逊 Redshift 集群，请参阅 使用亚马逊 Redshift 数据 API 与亚马逊 Redshift 集群进行交互 。

介绍数据 API

Data API 使您能够使用所有类型的传统、云原生和容器化基于无服务器 Web 服务的应用程序和事件驱动的应用程序无缝访问来自 Redshift Serverless 的数据。

下图说明了这种架构。

数据 API 简化了 亚马逊云科技 开发工具包 支持的编程语言和平台（ 例如 Python、Go、Java、Node.js、PHP、Ruby 和 C++）的数据访问、提取和输出。

数据 API 无需配置驱动程序和管理数据库连接，从而简化了对 Amazon Redshift 的访问。取而代之的是，你只需调用数据 API 提供的安全 API 端点即可向 Redshift Serverless 运行 SQL 命令。Data API 负责管理数据库连接和缓冲数据。数据 API 是异步的，因此您可以稍后检索结果。您的查询结果将存储 24 小时。数据 API 联合了 亚马逊云科技 身份和访问管理 (IAM) 证书，因此你可以使用 Okta 或 Azure Active Directory 等身份提供商或存储在密钥管理器中的数据库证书，而无需在 API 调用中传递数据库证书。

对于使用 亚马逊云科技 L am bda 的客户，数据 API 提供了一种安全的方式来访问您的数据库，而无需为在 Amazon VPC 中启动 Lambda 函数带来额外开销。与 亚马逊云科技 SDK 的集成提供了一个编程接口，用于运行 SQL 语句和异步检索结果。

相关用例

Data API 不能取代 JDBC 和 ODBC 驱动程序，适用于不需要持续连接到无服务器数据仓库的用例。它适用于以下用例：

• 使用 亚马逊云科技 开发工具包支持的任何编程语言从自定义应用程序访问 Amazon Redshift。这使您能够集成基于网络服务的应用程序，使用运行 SQL 语句的 API 访问来自 Amazon Redshift 的数据。例如，你可以通过 JavaScript 运行 SQL。
• 构建无服务器数据处理工作流程。
• 设计异步 Web 仪表板是因为 Data API 允许您运行长时间运行的查询，而不必等待查询完成。
• 运行一次查询并多次检索结果，而不必在 24 小时内再次运行查询。
• 使用 亚马逊云科技 Step Fun ctions 、Lambda 和存储过程构建 ETL 管道。
• 简化了从亚马逊 S ageMaker 和 Jupyter 笔记本电脑访问亚马逊 Red shift 的过程。
• 使用 亚马逊 EventB ridge 和 L ambda 构建事件驱动型应用程序。
• 调度 SQL 脚本以简化实例化视图的数据加载、卸载和刷新。

数据 API GitHub 存储库 为 Redshift 无服务器集群和预配置集群的不同用例 提供了示例。

创建 Redshift 无服务器工作组

如果你尚未创建 Redshift 无服务器数据仓库，或者想要创建一个新的数据仓库，请参阅 入门指南 。 本指南将引导您完成创建命名空间和工作组的步骤，并将名称设置为默认名称。此外，请确保您已创建一个 IAM 角色，并确保您附加到 Redshift 无服务器命名空间的 IAM 角色具有 A mazon s3ReadonlyAccess 权限。您可以使用 亚马逊云科技 管理控制台 创建 IAM 角色并分配 A mazon Simple Storage Servic e (Amazon S3) 权限（请参阅 从 Amazon S3 加载数据 ）。在这篇文章中，我们使用 COPY 命令创建表并加载数据。

使用数据 API 的先决条件

您必须获得访问数据 API 的授权。亚马逊 Redshift 提供 reds hiftDataFullAccess 管理策略，该策略提供对数据 API 的完全访问权限 。该策略还允许访问 Redshift 无服务器工作组、密钥管理器以及使用 IAM 证书对 Redshift 无服务器工作组进行身份验证和访问所需的 API 操作。

您还可以创建自己的 IAM 策略，该策略允许访问特定资源，方法是使用 redshiftDataFullA ccess 作为模板开始。

数据 API 允许您使用您的 IAM 证书或存储在 Secrets Manager 中的密钥访问您的数据库。在这篇文章中，我们使用 IAM 证书。

当您联合您的 IAM 证书以连接到 Amazon Redshift 时，它会自动为正在使用的 IAM 用户创建一个数据库用户。它使用 getCredit ials API 来获取临时数据库证书。如果您想通过此 API 向用户提供特定的数据库权限，则可以使用标签名为 RedshiftdBrole s 的 IAM 角色，并使用以冒号分 隔的角色列表。例如，如果要分配销售和分析师 等 数据库角色 ，则可以为 RedshiftdBroles 分配价值 s ales: an alyst。

使用 亚马逊云科技 CLI 中的数据 API

您可以使用 亚马逊云科技 CLI 中的数据 API 与 Redshift 无服务器工作组和命名空间进行交互。有关配置 亚马逊云科技 CLI 的说明，请参阅 设置 亚马逊云科技 CLI 。 亚马逊 Redshift 无服务器命令行界面 （aw s redshift-serverless ）是 AWS CLI 的一部分，它允许您管理 Amazon Redshift 工作组和命名空间，例如创建、删除、设置使用限制、标记资源等。数据 API 为 亚马逊云科技 CLI（ aws redshi ft-data ）提供了命令行接口，允许您与 Redshift S erverless 中的数据库进行交互。

您可以使用以下命令调 用 帮助：

aws redshift-data help

下表显示了 Data API CLI 中可用的不同命令。

如果你想获得有关特定命令的帮助，请运行以下命令：

aws redshift-data list-tables help

现在我们来看看如何使用这些命令。

列出数据库

大多数组织在其亚马逊 Redshift 工作组中使用单一数据库。您可以使用以下命令列出无服务器端点中的数据库。此操作需要您连接到数据库，因此需要数据库凭据。

aws redshift-data list-databases --database dev --workgroup-name default

列出架构

与列出数据库类似，您可以使用 list- schemas 命令列出架构：

aws redshift-data list-schemas --database dev --workgroup-name default

如果您有多个与 演示相匹配的架构 （ dem o 、d emo2 、demo3 等），则可以选择提供一种模式来筛选与该模式匹配的结果：

aws redshift-data list-schemas --database dev --workgroup-name default --schema-pattern "demo%"

列出表格

数据 API 提供了一个简单的命令， 即列表表 ，用于列出数据库中的表。一个架构中可能有数千个表；Data API 允许您对结果集进行分页或通过提供筛选条件来筛选表列表。

您可以使用 表模式在架构中进行搜索 ；例如，您可以按表名前缀筛选数据库中所有架构的表列表，或者使用架构模式以特定的架构模式筛选表列表。

以下是同时使用两者的代码示例：

aws redshift-data list-tables --database dev --workgroup-name default --schema-pattern "demo%" --table-pattern “orders%”

运行 SQL 命令

您可以使用数据 API 运行 Amazon Redshift 的 SELECT、DML、DDL、COPY 或卸载命令。如果您想在查询运行后向 EventBridge 发送 事件，则可以选择指定 —with- event 选项，然后 Data API 将发送带有 quer yID 和最终运行状态的事件。

创建架构

让我们使用 Data API 来看看如何创建架构。以下命令允许您在数据库中创建架构。如果您已经预先创建了架构，则不必运行此 SQL。必须指定 —-sql 才能指定 SQ L 命令。

aws redshift-data execute-statement --database dev --workgroup-name default \
--sql "CREATE SCHEMA demo;"

以下显示了 执行语句的示例输出：

{
    "CreatedAt": "2023-04-07T17:14:43.038000+00:00",
    "Database": "dev",
    "DbUser": "IAMR:Admin",
    "Id": "8e4e5af3-9af9-4567-8e70-7849515b3a79",
    "WorkgroupName": "default"
}

我们将在本文后面讨论如何检查使用 执行语句运行的 SQL 的状态。

创建表

您可以使用以下命令通过 CLI 创建表：

aws redshift-data execute-statement --database dev --workgroup-name default  \
   --sql "CREATE TABLE demo.green_201601( \
  vendorid                VARCHAR(4), \
  pickup_datetime         TIMESTAMP, \
  dropoff_datetime        TIMESTAMP, \
  store_and_fwd_flag      VARCHAR(1), \
  ratecode                INT, \
  pickup_longitude        FLOAT4, \
  pickup_latitude         FLOAT4, \
  dropoff_longitude       FLOAT4, \
  dropoff_latitude        FLOAT4, \
  passenger_count         INT, \
  trip_distance           FLOAT4, \
  fare_amount             FLOAT4, \
  extra                   FLOAT4, \
  mta_tax                 FLOAT4, \
  tip_amount              FLOAT4, \
  tolls_amount            FLOAT4, \
  ehail_fee               FLOAT4, \
  improvement_surcharge   FLOAT4, \
  total_amount            FLOAT4, \
  payment_type            VARCHAR(4),\
  trip_type               VARCHAR(4));" 

加载样本数据

使用 COPY 命令可以将批量数据加载到 Amazon Redshift 中的表中。您可以使用以下命令将数据加载到我们之前创建的表中：

aws redshift-data execute-statement --database dev --workgroup-name default --sql "COPY demo.green_201601 \
FROM 's3://us-west-2.serverless-analytics/NYC-Pub/green/green_tripdata_2016-01' \
IAM_ROLE default \
DATEFORMAT 'auto' \
IGNOREHEADER 1 \
DELIMITER ',' \
IGNOREBLANKLINES \
REGION 'us-west-2';" 

检索数据

以下查询使用了我们之前创建的表：

aws redshift-data execute-statement --database dev --workgroup-name default --sql "SELECT ratecode,  \
COUNT(*) FROM demo.green_201601 WHERE \
trip_distance > 5 GROUP BY 1 ORDER BY 1;"

以下显示了示例输出：

{
    "CreatedAt": "2023-04-07T17:25:16.030000+00:00",
    "Database": "dev",
    "DbUser": "IAMR:Admin",
    "Id": "cae88c08-0bb4-4279-8845-d5a8fefafade",
    "WorkgroupName": "default"
}

您可以使用作为 执行语句输出收到的语句 ID 来获取结果。

检查语句的状态

你可以使用 描述语句来检查你的陈述的状态。 d escribe-stat ement 的输出 提供了其他详细信息，例如 PID、查询时长、结果集的行数和大小，以及 Amazon Redshift 给出的查询 ID。你必须指定运行 执行语句命令时得到的语句 ID 。参见以下命令：

aws redshift-data describe-statement --id cae88c08-0bb4-4279-8845-d5a8fefafade \

以下是示例输出：

{
     "CreatedAt": "2023-04-07T17:27:15.937000+00:00",
     "Duration": 2602410468,
     "HasResultSet": true,
     "Id": "cae88c08-0bb4-4279-8845-d5a8fefafade",
     "QueryString": " SELECT ratecode, COUNT(*) FROM 
     demo.green_201601 WHERE
     trip_distance > 5 GROUP BY 1 ORDER BY 1;",
     "RedshiftPid": 1073815670,
     "WorkgroupName": "default",
     "UpdatedAt": "2023-04-07T17:27:18.539000+00:00"
}

语句的状态可以是 “开始”、“完成”、“中止” 或 “失败”。

运行带参数的 SQL 语句

您可以运行带参数的 SQL 语句。以下示例在 SQL 中使用两个使用名称-值对指定的命名参数：

aws redshift-data execute-statement --database dev --workgroup-name default --sql "select sellerid,sum(pricepaid) totalsales from sales where eventid >= :eventid and sellerid > :selrid group by sellerid"  --parameters "[{\"name\": \"selrid\", \"value\": \"100\"},{\"name\": \"eventid\", \"value\": \"100\"}]"

描述语句 返回 queryParameters 以及 queryString。

您可以将参数列表中的名称-值对映射到 SQL 文本中的一个或多个参数，并且名称-值参数可以按随机顺序排列。您不能将 NULL 值或零长度值指定为参数。

取消正在运行的语句

如果您的查询仍在运行，则可以使用 取消语句 取消 S QL 查询。参见以下命令：

aws redshift-data cancel-statement --id 39a0de2f-e85e-45ff-a0d7-cd074c348120

从您的查询中获取结果

你可以使用 get-statement -result 来获取 查询结果。查询结果将存储 24 小时。参见以下命令：

aws redshift-data get-statement-result --id 7b61da88-1b11-4ade-956a-21085a29118d

结果的输出包含元数据，例如提取的记录数、列元数据和分页标记。

运行多条 SQL 语句

您可以使用数据 API 在一次事务中为 Amazon Redshift 运行多个 SELECT、DML、DDL、COPY 或 UNLOAD 命令。 批处理执行语句使您 能够创建表并运行多个 COPY 命令，或者创建临时表作为报告系统的一部分，并对该临时表运行查询。参见以下代码：

aws redshift-data batch-execute-statement --database dev --workgroup-name default \
--sqls "create temporary table mysales \
(firstname, lastname, total_quantity ) as \
SELECT firstname, lastname, total_quantity \
FROM   (SELECT buyerid, sum(qtysold) total_quantity \
        FROM  sales  \
        GROUP BY buyerid \
        ORDER BY total_quantity desc limit 10) Q, users \
WHERE Q.buyerid = userid \ 
ORDER BY Q.total_quantity desc;" "select * from mysales limit 100;"

多 语句查询的 des cribe-statement 显示所有子语句的状态：

{

{
"CreatedAt": "2023-04-10T14:01:11.257000-07:00",
"Duration": 30564173,
"HasResultSet": true,
"Id": "23d99d7f-fd13-4686-92c8-e2c279715c21",
"RedshiftPid": 1073922185,
"RedshiftQueryId": 0,
"ResultRows": -1,
"ResultSize": -1,
"Status": "FINISHED",
"SubStatements": [
{
"CreatedAt": "2023-04-10T14:01:11.357000-07:00",
"Duration": 12779028,
"HasResultSet": false,
"Id": "23d99d7f-fd13-4686-92c8-e2c279715c21:1",
"QueryString": "create temporary table mysales (firstname, lastname,
total_quantity ) as \nSELECT firstname, lastname, total_quantity \nFROM (SELECT
buyerid, sum(qtysold) total_quantity\nFROM sales\nGROUP BY
buyerid\nORDER BY total_quantity desc limit 10) Q, users\nWHERE Q.buyerid =
userid\nORDER BY Q.total_quantity desc;",
"RedshiftQueryId": 0,
"ResultRows": 0,
"ResultSize": 0,
"Status": "FINISHED",
"UpdatedAt": "2023-04-10T14:01:11.807000-07:00"
},
{
"CreatedAt": "2023-04-10T14:01:11.357000-07:00",
"Duration": 17785145,
"HasResultSet": true,
"Id": "23d99d7f-fd13-4686-92c8-e2c279715c21:2",
"QueryString": ""select *\nfrom mysales limit 100;",
"RedshiftQueryId": 0,
"ResultRows": 40,
"ResultSize": 1276,
"Status": "FINISHED",
"UpdatedAt": "2023-04-10T14:01:11.911000-07:00"
}
],
"UpdatedAt": "2023-04-10T14:01:11.970000-07:00",
"WorkgroupName": "default"
}

在前面的示例中，我们有两条 SQL 语句，因此输出包括 SQL 语句的 ID，分别为 23d99d7f-fd13-4686-92c8-e2c279715c 21:1 和 23d99d7f-fd13-4686-92c8-e2c279715c 21 :2。 批处理 SQL 语句的每个子语句都有一个状态，并且批处理语句的状态使用最后一个子语句的状态进行更新。例如，如果最后一条语句的状态为 FAILED，则批处理语句的状态显示为 FAILED。

您可以单独获取每条语句的查询结果。在我们的示例中，第一条语句是用于创建临时表的 SQL 语句，因此第一条语句没有结果可供检索。您可以通过提供子语句的语句 ID 来检索第二条语句的结果集：

aws redshift-data get-statement-result --id 23d99d7f-fd13-4686-92c8-e2c279715c21:2

将数据 API 与密钥管理器一起使用

数据 API 允许您使用存储在密钥管理器中的数据库凭据。您可以将密钥类型创建为其他类型的密钥，然后指定用户名和密码。请注意，你无法选择 Amazon Redshift 集群，因为 Redshift 无服务器与集群不同。

假设你以 Defa ul Twg 的名义为证书创建了一个密钥。你可以使用 secret-arn 参数来传递你的密钥，如下所示：

aws redshift-data list-tables --database dev --workgroup-name default --secret-arn defaultWG --region us-west-1

导出数据

Amazon Redshift 允许您使用带有 SELECT 语句的 UNLOAD 命令从数据库表导出到 S3 存储桶中的一组文件。您可以卸载文本或 Parquet 格式的数据。以下命令向您展示了如何在 Data API 中使用数据湖导出的示例：

aws redshift-data execute-statement --database dev --workgroup-name default --sql "unload ('select * from demo.green_201601') to '<your-S3-bucket>' iam_role '<your-iam-role>'; " 

如果要在 UN LOAD 中使用多个语句或将 UNLOAD 与其他 SQL 语句结合使用，则可以使用批处理执行语句。

使用 亚马逊云科技 开发工具包中的数据 API

您可以在 亚马逊云科技 开发工具包支持的任何编程语言中使用数据 API。在这篇文章中，我们使用适用于 Python 的 亚马逊云科技 开发工具包 (Boto3) 作为示例来说明数据 API 的功能。

我们首先导入 Boto3 包并建立一个会话：

import botocore.session as bc
import boto3

def get_client(service, endpoint=None, region="us-west-2"):
    session = bc.get_session()
    s = boto3.Session(botocore_session=session, region_name=region)
    if endpoint:
        return s.client(service, endpoint_url=endpoint)
    return s.client(service)

获取客户端对象

你可以从 boto3.Session 对象中使用 Redshift Data 创建客户端 对象：

rsd = get_client('redshift-data')

如果你不想创建会话，你的客户端就像下面的代码一样简单：

import boto3
client = boto3.client('redshift-data')

运行语句

以下示例代码使用密钥管理器密钥运行语句。在这篇文章中，我们使用之前创建的表。您可以在 SQL 参数中使用 DDL、DML、COPY 和 UNLOAD：

resp = rsd.execute_statement(
    WorkgroupName ="default",
Database = "dev",
Sql = "SELECT ratecode, COUNT(*) totalrides FROM demo.green_201601 WHERE trip_distance > 5 GROUP BY 1 ORDER BY 1;" 
)

正如我们前面所讨论的，运行查询是异步的；运行语句会返回 executeStatementOut put ，其中包括语句 ID。

如果要在语句完成后 将事件发布到 EventBridge，则可以使用附加参数，将 Event 设置为 true：

resp = rsd.execute_statement(
    Database="dev",
    WorkgroupName="default",
    Sql="SELECT ratecode, COUNT(*) totalrides FROM demo.green_201601 WHERE trip_distance > 5 GROUP BY 1 ORDER BY 1;",
WithEvent=True
)

描述陈述

你可以使用 desc ribe_stat ement 来查找查询的状态和检索到的记录数：

id=resp['Id']
desc = rsd.describe_statement(Id=id)
if desc["Status"] == "FINISHED":
    print(desc["ResultRows"])

从您的查询中获取结果

如果您的查询已完成，则可以使用 get_statement_resul t 来检索查询结果：

if desc and desc["ResultRows"]  > 0:
    result = rsd.get_statement_result(Id=qid)

get_statement_resul t 命令返回一个 JSON 对象，其中包含结果和实际结果集的元数据。如果要以用户友好的格式显示结果，则可能需要处理数据以格式化结果。

获取和格式化结果

在这篇文章中，我们演示了如何使用 Pandas 框架格式化结果。 post_process 函数处理元数据和结果以填充 DataFrame。查询函数从 Amazon Redshift 集群中的数据库中检索结果。参见以下代码：

import pandas as pd

def post_process(meta, records):
    columns = [k["name"] for k in meta]
    rows = []
    for r in records:
        tmp = []
        for c in r:
            tmp.append(c[list(c.keys())[0]])
        rows.append(tmp)
    return pd.DataFrame(rows, columns=columns)

def query(sql, workgroup="default ", database="dev"):
    resp = rsd.execute_statement(
        Database=database,
        WorkgroupName=workgroup,
        Sql=sql
    )
    qid = resp["Id"]
    print(qid)
    desc = None
    while True:
        desc = rsd.describe_statement(Id=qid)
        if desc["Status"] == "FINISHED" or desc["Status"] == "FAILED":
            break
    	print(desc["ResultRows"])
    if desc and desc["ResultRows"]  > 0:
        result = rsd.get_statement_result(Id=qid)
        rows, meta = result["Records"], result["ColumnMetadata"]
        return post_process(meta, rows)

pf=query("select * from demo.customer_activity limit 100;")
print(pf)

在这篇文章中，我们演示了在 Python 中使用数据 API 和 Redshift 无服务器。但是，您可以将数据 API 与 亚马逊云科技 开发工具包支持的其他编程语言一起使用。你可以阅读罗氏如何 使用带有谷歌表格的数据 API 来实现亚马逊 Redshift 数据的访问 民主化 。 您也可以使用 Redshift 无服务器来解决此类用例。

最佳实践

在使用 Data API 时，我们建议采用以下最佳做法：

• 将您的 IAM 证书联合到数据库以连接亚马逊 Redshift。Redshift 无服务器允许用户使用 getCreditials 获取临时数据库证书。 Redshift 无服务器将访问范围限定为特定 IAM 用户，数据库用户是自动创建的。
• 如果您不希望用户使用临时证书，请使用自定义策略在生产环境中提供对数据 API 的精细访问。在此类用例中，你必须使用密钥管理器来管理你的证书。
• 不要从您的客户端检索大量数据，并使用 UNLOAD 命令将查询结果导出到 Amazon S3。使用数据 API，您只能检索 100 MB 的数据。
• 不要忘记在 24 小时内检索结果；结果仅存储 24 小时。

结论

在这篇文章中，我们介绍了如何在 Redshift 无服务器中使用数据 API。我们还演示了如何使用 Amazon Redshift CLI 中的数据 API，以及如何使用 亚马逊云科技 开发工具包使用 Python。此外，我们还讨论了使用数据 API 的最佳实践。

要了解更多信息，请参阅 使用亚马逊 Redshift 数据 API 或访问数据 API GitHub 存储库 获取代码示例。

作者简介

Debu Pand a 是 亚马逊云科技 产品管理高级经理，是分析、应用程序平台和数据库技术领域的行业领导者，在 IT 领域拥有超过 25 年的经验。Debu 发表了许多关于分析、企业 Java 和数据库的文章，并在诸如 re: Invent、Oracle Open World 和 Java One 等多个会议上发表过演讲。他是《EJB 3 在行动》（曼宁出版社 2007、2014 年）和《中间件管理》（Packt）的主要作者。

彭飞是一名软件开发工程师，在亚马逊 Redshift 团队 工作。