Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

亚马逊 Data Firehose 以前被称为亚马逊 Kinesis Data Firehose

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

在 Firehose 中转换您的输入记录格式

在将数据存储到亚马逊 S3 之前，Amazon Data Firehose 可以将您的输入数据的格式从 JSON 转换为 Apache Parq uet 或 Apache ORC。Parquet 和 ORC 是列式数据格式，与 JSON 等行式格式相比，其可节省空间并更快地启用查询。如果您想要转换 JSON 以外的输入格式，如逗号分隔值（CSV）或结构化文本，您可以先使用 Amazon Lambda 来将其转换为 JSON 格式。有关更多信息，请参阅 亚马逊数据 Firehose 数据转换。

记录格式转换要求

Amazon Data Firehose 需要以下三个元素来转换记录数据的格式：

即使您在将记录发送到 Amazon Data Firehose 之前汇总了这些记录，也可以转换数据的格式。

选择 JSON 解串器

SerDe如果您的输入 JSO N 包含以下格式的时间戳，请选择 OpenX JSON：

OpenX JSON SerDe 可以将句点 (.) 转换为下划线 (_)。它还可以在对 JSON 键进行反串行化前将其转换为小写。有关此反序列化器通过 Amazon Data Firehose 提供的选项的更多信息，请参阅 OpenX。JsonSerDe

如果你不确定要选择哪个反序列化器，请使用 OpenX JSON SerDe，除非你有它不支持的时间戳。

如果您的时间戳格式与之前列出的格式不同，请使用 Apache Hiv e JSON。 SerDe选择此解串器后，您可以指定要使用的时间戳格式。为此，请遵循 Joda-Time DateTimeFormat 格式字符串的模式语法。有关更多信息，请参阅类 DateTimeFormat。

您还可以使用特殊值 millis 来解析时间戳（毫秒，以纪元格式表示）。如果您未指定格式，Amazon Data Firehose 会java.sql.Timestamp::valueOf默认使用该格式。

Hive JSON SerDe 不允许执行以下操作：

Hive SerDe 不会将嵌套的 JSON 转换为字符串。例如，如果您有 {"a":{"inner":1}}，它不会将 {"inner":1} 视为字符串。

选择串行化器

选择的串行化器取决于您的业务需求。要了解有关这两个序列化器选项的更多信息，请参阅 ORC SerDe 和 P arquet。 SerDe

转换输入记录格式（控制台）

创建或更新 Firehose 直播时，可以在控制台上启用数据格式转换。启用数据格式转换后，Amazon S3 是您可以为 Firehose 数据流配置的唯一目标。此外，启用格式转换时，系统将禁用 Amazon S3 压缩。但是，Snappy 压缩会作为自动转换过程的一部分自动进行。Amazon Data Firehose 在本例中使用的 Snappy 框架格式与 Hadoop 兼容。这意味着，您可以使用 Snappy 压缩的结果并在 Athena 中对这些数据运行查询。有关 Hadoop 所依赖的 Snappy 取景格式，请参阅.java。BlockCompressorStream

转换输入记录格式 (API)

如果你想让 Amazon Data Firehose 将你的输入数据格式从 JSON 转换为 Parquet 或 ORC，请在 extendedS3 或 Extended DestinationConfiguration S DataFormatConversionConfiguration3 中指定可选元素。DestinationUpdate如果您指定 DataFormatConversionConfiguration，则适用以下限制：

记录格式转换错误处理

当 Amazon Data Firehose 无法解析或反序列化记录时（例如，当数据与架构不匹配时），它会使用错误前缀将其写入 Amazon S3。如果此写入失败，Amazon Data Firehose 将永远重试，从而阻止进一步交付。对于每条失败的记录，Amazon Data Firehose 都会使用以下架构编写一个 JSON 文档：

{
  "attemptsMade": long,
  "arrivalTimestamp": long,
  "lastErrorCode": string,
  "lastErrorMessage": string,
  "attemptEndingTimestamp": long,
  "rawData": string,
  "sequenceNumber": string,
  "subSequenceNumber": long,
  "dataCatalogTable": {
    "catalogId": string,
    "databaseName": string,
    "tableName": string,
    "region": string,
    "versionId": string,
    "catalogArn": string
  }
}

记录格式转换示例

有关如何使用设置记录格式转换的示例Amazon CloudFormation，请参阅Amazon::DataFirehose:: DeliveryStream。