从 REST API 读取数据

注册 REST API ConnectionType 并创建 AWS Glue 连接后，您可以在 AWS Glue ETL 作业中从 REST API 读取数据。通过此连接，您可以在同一作业中将外部 REST API 数据与其他数据来源一起处理。您需要连接名称和实体名称才能读取数据。

以下示例演示如何使用 Python 从 REST API 数据来源读取数据：

rest_read = glueContext.create_dynamic_frame.from_options(
    connection_type="rest",
    connection_options={
        "connectionName": "connection-name",
        "ENTITY_NAME": "entity-name",
        "CONNECTION_TYPE": "REST-connection-type"
    }
)

筛选数据

您可以将筛选条件谓词下推至源 REST API，以减少传输的数据量。REST API 连接器支持两种筛选模式：

要在 AWS Glue ETL 作业中应用筛选条件谓词，请使用 FILTER_PREDICATE 连接选项：

rest_read = glueContext.create_dynamic_frame.from_options(
    connection_type="rest",
    connection_options={
        "connectionName": "connection-name",
        "ENTITY_NAME": "entity-name",
        "CONNECTION_TYPE": "REST-connection-type",
        "FILTER_PREDICATE": "status = \"ACTIVE\" AND lastUpdated >= \"2024-01-01T00:00:00.000Z\""
    }
)

FilterConfiguration 属性

使用 FilterConfiguration 对象在 ConnectionType 注册中配置筛选行为。下表描述了可用的属性：

FilterStringConfiguration properties

下表描述了 FILTER_STRING 模式的属性：

BetweenConfiguration 属性

下表描述了 BETWEEN 处理属性：

支持的运算符

筛选条件谓词中支持以下运算符：EQUAL_TO、GREATER_THAN、LESS_THAN、GREATER_THAN_OR_EQUAL_TO、LESS_THAN_OR_EQUAL_TO、NOT_EQUAL_TO、CONTAINS、BETWEEN、AND、OR。

字段级覆盖

您可以使用架构定义中的 FilterOverrides 配置每个字段的筛选行为。以下覆盖属性可用：

示例：QUERY_PARAMS 模式

以下示例显示了 REST API 的 FilterConfiguration，该 API 使用括号符号表示操作符的查询参数和 epoch 时间戳表示日期值：

"FilterConfiguration": {
    "FilterMode": "QUERY_PARAMS",
    "OperatorMappings": {
        "EQUAL_TO": "{FIELD}",
        "GREATER_THAN_OR_EQUAL_TO": "{FIELD}[gte]",
        "LESS_THAN_OR_EQUAL_TO": "{FIELD}[lte]"
    },
    "DateTimeFormat": "EPOCH_SECONDS",
    "BetweenConfiguration": {
        "LowBoundKey": "{FIELD}[gte]",
        "HighBoundKey": "{FIELD}[lte]"
    }
}

使用此配置，created >= 2024-01-01 AND created <= 2024-06-01 的输入筛选条件将生成 URL 查询字符串：?created[gte]=1704067200&created[lte]=1717200000

示例：FILTER_STRING 模式

以下示例显示了 REST API 的 FilterConfiguration，该 API 使用带有空格填充运算符的单个筛选字符串参数：

"FilterConfiguration": {
    "FilterMode": "FILTER_STRING",
    "OperatorMappings": {
        "EQUAL_TO": " eq ",
        "GREATER_THAN": " gt ",
        "GREATER_THAN_OR_EQUAL_TO": " ge ",
        "LESS_THAN": " lt ",
        "AND": " and ",
        "OR": " or "
    },
    "DateTimeFormat": "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'",
    "BetweenConfiguration": {
        "Template": "{FIELD} ge {LOW} and {FIELD} le {HIGH}"
    },
    "FilterStringConfiguration": {
        "FilterStringKey": "search",
        "QuoteStringValues": true,
        "QuoteCharacter": "\""
    }
}

使用此配置，status = "ACTIVE" AND lastUpdated > 2024-01-01T00:00:00.000Z 的输入筛选条件将生成 URL 查询字符串：?search=status eq "ACTIVE" and lastUpdated gt "2024-01-01T00:00:00.000Z"

对查询进行分区

您可以将数据读取拆分为跨 Spark 个工作线程的并行分区，以提高吞吐量。REST API 连接器支持基于字段的分区，它根据指定的字段将数据划分为范围。

Spark 作业参数

以下连接选项控制分区行为：

基于字段的分区

基于字段的分区根据指定字段将数据划分为多个范围。以下示例基于 lastUpdated 字段对数据进行分区：

rest_read = glueContext.create_dynamic_frame.from_options(
    connection_type="rest",
    connection_options={
        "connectionName": "connection-name",
        "ENTITY_NAME": "entity-name",
        "CONNECTION_TYPE": "REST-connection-type",
        "PARTITION_FIELD": "lastUpdated",
        "LOWER_BOUND": "2024-01-01T00:00:00.000Z",
        "UPPER_BOUND": "2024-12-31T00:00:00.000Z",
        "NUM_PARTITIONS": "4"
    }
)

在 lastUpdated 字段上设置 4 个分区时，工作分布如下：

在 RegisterConnectionType 中配置分区支持

要启用基于字段的分区，请在架构定义中将字段标记为可分区：

"Schema": {
    "lastUpdated": {
        "FieldDataType": "TIMESTAMP",
        "IsPartitionable": true
    }
}

以下 FieldDataType 值支持分区：