在 Amazon Data Firehose 上使用 Apache Iceberg 表

Apache Iceberg 是一种用于执行大数据分析的高性能开源表格格式。Apache Iceberg 为亚马逊 S3 数据湖带来了SQL表格的可靠性和简单性，并使 Spark、Flink、Trino、Hive 和 Impala 等开源分析引擎可以同时处理相同的数据。有关 Apache Iceberg 的更多信息，请参阅。https://iceberg.apache.org/

你可以使用 Firehose 将流数据直接传送到亚马逊 S3 中的 Apache Iceberg Tables。使用此功能，您可以将记录从单个流路由到不同的 Apache 冰山表，并自动对 Apache Iceberg Tables 中的记录应用插入、更新和删除操作。此功能需要使用 Amazon Glue Data Catalog。

支持的区域和数据类型

先决条件

在开始之前，请完成以下先决条件。

设置直播

要创建 Firehose 直播，您需要配置以下内容。

来源和目的地

要向 Apache Iceberg Tables 传送数据，请为您的直播选择来源。目前，我们不支持将亚马逊MSK作为向 Apache Iceberg Tables 传输直播的来源。

如果您选择 Amazon Kinesis Data Streams 作为来源，请在控制台的 “来源” 设置下，浏览或创建 Kinesis Data ARN Streams 或者按格式输入数据流。arn:aws:kinesis:[Region]:[AccountId]:stream/[StreamName]

接下来，选择 Apache Iceberg Table s 作为目标并提供 Firehose 直播名称。

数据转换

要对数据执行自定义转换，例如在传入流中添加或修改记录，您可以将 Lambda 函数添加到您的 Firehose 流中。有关在 Firehose 流中使用 Lambda 进行数据转换的更多信息，请参阅。在 Amazon Data Firehose 中转换数据

对于 Apache Iceberg Tables，您需要指定如何将传入流中的数据映射到目标中的特定表和列。您还需要指定如何将传入记录路由到不同的目标表。您可以通过以下方式之一执行此操作。

有关更多信息，请参阅 格式化输入记录。

数据目录

Apache Iceberg 需要数据目录来指定表和列的定义。Firehose 与 Apache Iceber Amazon Glue Data Catalog g Tables 集成。如果您没有现有的数据目录，请参阅 Glu e 数据 Amazon 目录入门。

如果你使用 Hive Metastore，请参阅将数据目录连接到外部 Hive 元数据仓，了解如何连接 Hive Metastor Amazon Glue Data Catalog e。

您可以在与您的 Firehose 直播相同的账户和区域 Amazon Glue Data Catalog 中使用（默认），也可以从控制台的Amazon Web Services 区域下拉列表中选择不同的区域。

唯一密钥

唯一键是源记录中的一个或多个字段，用于唯一标识 Apache Iceberg Tables 中的一行。在源记录中使用唯一密钥是可选的。如果要对某些表执行更新和删除操作，则需要在源记录中具有进入这些表的唯一密钥，并且必须在 Firehose 流创建过程中配置这些唯一密钥。

Retry duration

您可以使用此配置来指定 Firehose 在写入 Amazon S3 中的 Apache Iceberg 表时遇到失败时应尝试重试的时长（以秒为单位）。您可以将 0 到 7200 秒之间的任何值设置为执行重试。默认情况下，Firehose 会重试 300 秒。

配送或处理失败

您必须将 Firehose 配置为将记录传送到 S3 备份存储桶，以防它在重试持续时间到期后处理或传输数据流时遇到故障。为此，请在控制台的 Backup 设置中配置 S3 备份存储桶和 S3 备份存储桶错误输出前缀。

缓冲区提示

Firehose 会将内存中传入的流数据缓冲到一定大小（缓冲大小）和一段时间（缓冲间隔），然后再将其传送到 Apache Iceberg 表。您可以选择 1—128 的缓冲区大小 MiBs 和 0—900 秒的缓冲间隔。缓冲区提示越高，S3 写入次数越少，由于数据文件越大，压缩成本就会降低；查询执行速度更快，但延迟更高。较低的缓冲区提示值以较低的延迟提供数据。

高级设置

您可以为 Apache Iceberg Tables 配置服务器端加密、错误日志、权限和标签。有关更多信息，请参阅 配置高级设置。在 “高级设置” 中，在 “服务访问权限” 下，您需要添加作为其中一部分创建的IAM角色先决条件 。Firehose 将担任访问 Amazon Glue 表和写入亚马逊 S3 存储桶的角色。

Firehose 直播创建可能需要几分钟才能完成。有关任何与直播创建相关的错误，请参阅处理错误。成功创建 Firehose 流后，您可以开始向其中提取数据，并可以使用控制台查看 Apache Iceberg 表中的数据。 Amazon Glue

处理错误

直播错误

Apache Iceberg 传送错误

Firehose 将所有传送错误发送到 CloudWatch 日志和 S3 错误存储桶。

指标

为了向 Apache Iceberg Tables 传输数据，Firehose 会在直播级别发布以下 CloudWatch 指标。

考虑因素和限制

Firehose 对用于预览的 Apache Iceberg 表的支持有以下注意事项和限制。

格式化输入记录

你可以使用 Dire PUT PutRecord ct PutRecordBatch APIs 和/或 Amazon Kinesis Data Streams 将数据提取到 Firehose 中。 PutRecords API目前，对于这两个源，你都需要将你的JSON对象作为 blob 添加到 Data 字段中，如下例所示。

示例JSON对象

{
    "courseid": "5",
    "coursename": "Computer Science",
    "semester": "fall",
    "grade": "A"
 }

JSONas blob 已添加到输入记录的数据字段

{ 
    "Data": blob // byte array of JSON string as shown in previous sample 
 }

使用 Firehose 将流传输到 Apache Iceberg Tables 时，必须指定如何将传入的记录路由到不同的目标表以及要执行的操作。为此，Firehose 需要ADF_Metadat在您的直播记录中包含三个元数据字段，如以下示例所示。您还需要将您的添加JSON为的一部分ADF_Record，如以下示例所示。

{
    "ADF_Record": "{
    "courseid": "5",
    "coursename": "Computer Science",
    "semester": "fall",
    "grade": "A"
    }"
    "ADF_Metadata" : {
        "OTF_Metadata": { 
            "DestinationTableName": "iceberg_testing", //Required
            "DestinationDatabaseName": "course",// Required
            "Operation": "INSERT" // Optional (Possible values-INSERT, UPDATE, DELETE)
        }
    }
 }

在上面的示例中，DestinationDatabaseName是该记录的目标数据库的名称，并且是必填字段。 DestinationTableName是此记录的 Apache Iceberg Table 的名称，也是必填字段。 Operation是一个可选字段，可能的值为、INSERT（默认）UPDATE、和DELETE。根据您要对此记录执行的操作指定值。要执行UPDATE和DELETE操作，您必须在创建 Firehose 直播时为所需表配置唯一键。

按照上一个示例所示格式化数据后，可以将此JSON对象的 blob 添加到 Data 字段，如下所示。

{ 
    "Data": blob // byte array of JSON string shown in previous example
 }

您可以通过以下方式之一格式化流媒体记录。

import base64
import json

def lambda_handler(firehose_records_input, context):
    print("Received records for processing from DeliveryStream: " + firehose_records_input['deliveryStreamArn']
        + ", Region: " + firehose_records_input['region']
        + ", and InvocationId: " + firehose_records_input['invocationId'])

    # Create return value.
    firehose_records_output = {}
    # Create result object.
    firehose_records_output['records'] = []


    # Go through records and process them
    for firehose_record_input in firehose_records_input['records']:
        json_bytes = base64.b64decode(firehose_record_input['data']).decode('utf-8')
        original_data = json.loads(json_bytes)
        new_data = {'ADF_Record' : original_data}
        adf_metadata = json.loads('{"OTF_Metadata": {"DestinationTableName":"course", "DestinationDatabaseName": "iceberg_testing","Operation":"insert"}}')
        new_data["ADF_Metadata"] = adf_metadata
        firehose_record_output = {}

        # Must set proper record ID
        firehose_record_output['recordId'] = firehose_record_input['recordId']
        firehose_record_output['result'] =  'Ok'
        modified_json_bytes = json.dumps(new_data)
        firehose_record_output['data'] = base64.b64encode(modified_json_bytes.encode('utf-8'))
        firehose_records_output['records'].append(firehose_record_output)

    return firehose_records_output

import boto3
import json


delivery_stream_name = 'xxxxxx'
print("Putting data to " + delivery_stream_name)
firehose_client = boto3.client('firehose', region_name='eu-west-1')   # Replace 'region_name' with your AWS region

def load_from_local():
    records = []
    courseids = ["5"]
    coursenames = ["Computer Science"]
    for j in range(0, 1):
        # change here to any json you need to upload to iceberg
        raw_json = '{"courseid":"%s","coursename": "%s","semester":"fall","grade":"A"}' % (courseids[j], coursenames[j])
        print(raw_json)

        original_data = json.loads(raw_json)
        metadata = '{"OTF_Metadata": {"DestinationTableName":"course", "DestinationDatabaseName": "iceberg_testing","Operation":"insert"}}'
        adf_metadata = json.loads(metadata)

        iceberg_data = {'ADF_Record' : original_data}
        iceberg_data["ADF_Metadata"] = adf_metadata

        modified_json_bytes = json.dumps(iceberg_data)
        record = {'Data': modified_json_bytes}
        records.append(record)

    response = firehose_client.put_record_batch(DeliveryStreamName=delivery_stream_name, Records=records)
    print(response)

load_from_local()