亚马逊KDF与KDA：流式数据处理与分析的利器

### 亚马逊KDF与KDA：流式数据处理与分析的利器 #### 亚马逊KDF的特性与应用 亚马逊KDF（Kinesis Data Firehose）是数据管道中的重要组件，可与多个亚马逊及第三方服务和软件进行加密、网络、认证和授权操作。它能轻松从AWS IoT、Amazon CloudWatch Logs、Amazon CloudWatch Events和KDS等AWS服务中摄取数据，通过Lambda函数进行内联转换，并将数据输送至Amazon S3、Amazon Redshift、Amazon Elasticsearch、HTTP端点和其他第三方目的地。 ##### 复杂数据类型JSON键中的连字符问题 若传入的JSON记录在结构体中有带连字符的JSON键，Apache Hive会报错。由于KDF使用AWS Glue获取模式信息，而AWS Glue是与Apache Hive兼容的元存储，因此在数据格式转换时，若嵌套结构体的键中出现连字符，KDF也会报错。解决方法是修改AWS Glue中的模式，去除连字符或将其转换为下划线，然后利用OpenX JSON SerDe的列名到JSON键映射功能。 ##### Apache Hive JSON SerDe 该SerDe是Apache Hive的一部分，支持除OpenX JSON SerDe所列格式之外的时间戳格式。它允许使用Joda - Time的DateTimeFormat格式字符串的模式语法指定时间戳格式（https://www.joda.org/joda - time/apidocs/org/joda/time/format/DateTimeFormat.html）。若未指定格式，KDF默认使用java.sql.Timestamp::valueOf将表示时间戳的字符串对象转换为java.sql.Timestamp值。更多信息可参考https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#Languag eManualDDL - JSON。 ##### 模式配置 KDF需要AWS Glue数据目录中的模式位置。模式可以通过以下方式创建： 1. 使用AWS Glue控制台、CLI或SDK手动创建。 2. 先创建一个交付流，将少量示例记录以JSON格式交付到S3位置，然后使用AWS Glue爬虫爬取数据、推断模式，并在指定数据库的AWS Glue数据目录中创建具有推断模式的表。之后可创建新的交付流或更新现有交付流以启用数据格式转换，并使用AWS Glue爬虫创建的模式。若将爬虫指定的S3位置（即示例记录交付的位置）用作交付流的目标，可轻松使用AWS Glue数据目录表查询数据（先删除包含示例记录的对象）。爬虫还可按计划运行，随着时间推移添加更多数据和分区时，它能识别并向AWS Glue数据目录表添加额外分区。 ##### 序列化器 支持两种序列化器：Apache ORC和Apache Parquet。 ##### 数据格式转换错误处理 KDF交付流配置提供了指定S3错误前缀的选项，用于存放转换或数据格式转换失败的记录。若KDF遇到数据格式转换错误（如无法读取和解析记录、无法成功反序列化记录）、模式错误或因模式中不支持的结构导致的Hive错误，它会将记录与一些元数据一起写入S3错误前缀配置中指定的S3对象（文件夹）。若因权限问题或网络问题导致S3存储桶或前缀位置不可用，KDF会无限重试，可能导致后续记录交付停滞。由于KDF会存储传入记录24小时，若阻塞超过24小时，可能会造成数据丢失。为防止数据丢失并及时收到问题通知和采取纠正措施，必须监控KDF CloudWatch的SucceedConversion.Records、SucceedConversion.Bytes、FailedConversion.Records和FailedConversion.Bytes指标，并为FailedConversion.Records创建警报。 ##### KDF的监控指标 KDF与Amazon CloudWatch紧密集成，会向其发送错误消息和指标，这些指标有助于监控KDF的不同方面，可设置警报以在触发时接收通知或使用Lambda函数采取自动操作。以下是一些需要关注的指标： | 指标名称 | 含义 | | ---- | ---- | | IncomingBytes | 在指定时间段内成功摄取到交付流的字节数，可与生产者预期发送的字节数进行核对 | | IncomingRecords | 在指定时间段内成功摄取到交付流的记录数，可与生产者预期发送的记录数进行核对 | | IncomingPutRequests | 在指定时间段内成功的PutRecord和PutRecordBatch请求数 | | KinesisMillisBehindLatest | 当摄取源为KDS流时，该指标表示最后读取的记录落后于Kinesis数据流中最新记录的毫秒数，需监控并设置警报，以防KDF读取KDS流时落后超过保留期（默认24小时）导致数据丢失 | | RecordsPerSecondLimit | 表示KDF当前每秒记录数的限制，超过此限制可能会出现限流，适用于直接PUT摄取，可与IncomingRecords指标对比以确定何时请求增加KDF交付流的限制 | | BytesPerSecondLimit | 表示KDF当前每秒字节数的限制，超过此限制可能会出现限流，适用于直接PUT摄取，可与IncomingBytes指标对比以确定何时请求增加KDF交付流的限制 | | PutRequestsPerSecondLimit | 表示KDF当前每秒PUT请求（PutRecord和PutRecordBatch）的限制，超过此限制可能会出现限流，适用于直接PUT摄取，可与IncomingPutRequests指标对比以确定何时请求增加KDF交付流的限制 | | ThrottledRecords | 因数据摄取超过交付流限制而被限流的记录数，若该数值持续增加，需请求增加限制，建议为此指标创建警报 | | DeliveryTo<Destination>.Success | 适用于所有目的地，表示成功交付与失败交付的比率，需监控以识别失败情况 | | DeliveryTo<Destination>.DataFreshness | KDF中最旧记录的年龄，超过此年龄的记录已交付到目的地，指示交付到目的地的延迟，建议为此指标创建警报，当超过目的地配置中的指定缓冲限制时发出通知 | 此外，KDF还与AWS CloudTrail集成，启用后可记录针对KDF API的所有API调用，并将这些事件交付到S3存储桶，用于审计对KDF API的访问和操作。 #### 共享单车数据管道示例 以纽约多个自行车站的流式数据处理为例，说明KDF的应用。该用例是将来自自行车站的流式数据与存储在Amazon DynamoDB表中的地址信息结合，对数据进行缓冲和聚合，然后将其存入S3存储桶，形成数据湖，以便使用Apache Hive和Amazon Athena等大数据查询工具进行历史分析和洞察。 ##### 架构与配置 KDF交付流配置为使用KDS流作为数据源，通过Lambda转换为传入记录添加地址信息，然后使用KDF数据格式转换将传入记录转换为Parquet格式，使用AWS Glue数据目录中定义的模式。记录会缓冲15分钟或达到128 MB（以先到者为准），然后以Hive样式的分区格式将S3对象交付到指定的S3存储桶。 ##### 重现示例的步骤 1. **创建DynamoDB表**：使用以下AWS CLI命令创建用于存储自行车站地址的表。 ```bash aws dynamodb create-table \ --table-name BikeStationAddress \ --attri ```