要获得与亚马逊 Timestream 类似的功能 LiveAnalytics，可以考虑适用于 InfluxDB 的亚马逊 Timestream。适用于 InfluxDB 的 Amazon Timestream 提供简化的数据摄取和个位数毫秒级的查询响应时间，以实现实时分析。点击[此处](https://docs.amazonaws.cn/timestream/latest/developerguide/timestream-for-influxdb.html)了解更多信息。

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

# 将数据从 Amazon S3 摄取到适用于 InfluxDB 的 Timestream 自动化系统
<a name="data-ingestion-from-s3"></a>

 LiveAnalytics 导出的 Timestream 工具完成卸载过程后，自动化过程的下一步就开始了。此自动化使用 [InfluxDB 的导入工具](https://docs.influxdata.com/influxdb/v2/write-data/)，将数据传输到其专用的时间序列结构中。该过程对 Timestream 的数据模型进行转换，使其匹配 InfluxDB 的测量值、标签和字段概念。最后，该过程使用 [InfluxDB 的行协议](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/)高效加载数据。

完成迁移的工作流程分为四大阶段：

1. 使用 Timestream LiveAnalytics 导出工具卸载数据。

1. [数据转](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/targets/timestream_for_influxdb/transform/README.md)[换：使用 Amazon Athena 将 LiveAnalytics 数据的 Timestream 转换为 InfluxDB 线路协议格式（基于基数评估后定义的架构）。](https://docs.amazonaws.cn/athena/latest/ug/when-should-i-use-ate.html)

1. [数据摄取](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/targets/timestream_for_influxdb/ingestion/README.md)：将行协议数据集摄取到适用于 InfluxDB 的 Timestream 实例中。

1. [验证](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/targets/timestream_for_influxdb/validation/README.md)：（可选）您可以验证是否已摄取每个行协议点（数据转换步骤中需要 `--add-validation-field true`）。

**数据转换**

为了进行数据转换，我们开发了一个脚本，使用Amazon Athena将 LiveAnalytics 导出的数据镶木地板格式的时间流转换为InfluxDB的线路协议格式。Amazon Athena 提供无服务器查询服务，并以经济高效的方式处理海量时间序列数据，无需专用计算资源。

 脚本执行以下操作：
+ 将来自亚马逊 S3 存储桶 LiveAnalytics 的数据导出的 Timestream 加载到亚马逊 Athena 表中。
+ 执行数据映射，将存储在 Athena 表中的数据转换为行协议格式，并将其存储在 S3 存储桶中。

**数据映射**

下表显示了如何将 LiveAnalytics 数据的 Timestream 映射到线路协议数据。


| 概念的时间 LiveAnalytics 流 | 行协议概念 | 
| --- | --- | 
| [表名](https://docs.amazonaws.cn/timestream/latest/developerguide/API_Table.html) | [测量](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/#measurement) | 
| [Dimensions](https://docs.amazonaws.cn/timestream/latest/developerguide/API_Dimension.html) | [标签](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/#tag-set) | 
| [度量名称](https://docs.amazonaws.cn/timestream/latest/developerguide/data-modeling.html#data-modeling-measurenamemulti) | 标签（可选） | 
| [措施](https://docs.amazonaws.cn/timestream/latest/developerguide/API_MeasureValue.html) | [字段](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/#field-set) | 
| [时间](https://docs.amazonaws.cn//timestream/latest/developerguide/writes.html#writes.data-types) | [Timestamp](https://docs.influxdata.com/influxdb/v2/reference/syntax/line-protocol/#timestamp) | 

**先决条件和安装**

请参阅[转换脚本的自述文件](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/targets/timestream_for_influxdb/transform/README.md#prerequisites)中的“先决条件”和“安装”部分。

**用法**

要将存储在存储桶 example\_s3\_bucket 中的数据从 example\_database 中的 LiveAnalytics 表的时间流转换为 example\_table，请运行以下命令：

```
python3 transform.py \
        --database-name example_database \
        --tables example_table \
        --s3-bucket-path example_s3_bucket \
        --add-validation-field false
```

脚本完成后：
+ 在 Athena 中，将创建 example\_database\_example\_table 表，其中包含数据的时间流。 LiveAnalytics 
+ 在 Athena 中，将创建 lp\_example\_database\_example\_table 表，其中包含转换为线路协议点的数据的时间流。 LiveAnalytics 
+ 在 S3 存储桶 example\_s3\_bucket 中，将在路径 ` example_database/example_table/unload-<%Y-%m-%d-%H:%M:%S>/line-protocol-output` 中存储行协议数据。

**建议**

请参阅[转换脚本的自述文件](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/targets/timestream_for_influxdb/transform/README.md#prerequisites)，了解脚本的最新用法详情，以及后续迁移步骤（例如验证）需要这些输出。如果为提高基数而排除维度，请使用 `--dimensions-to-fields` 参数将特定维度更改为字段，从而调整架构以降低基数。

*添加用于验证的字段*

有关如何添加用于验证的字段的信息，请参阅转换脚本自述文件中的[添加用于验证的字段](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/targets/timestream_for_influxdb/transform/README.md#adding-a-field-for-validation)部分。

## 将数据摄取到适用于 InfluxDB 的 Timestream
<a name="data-ingestion-influxdb"></a>

InfluxDB 摄取脚本将压缩的行协议数据集摄取到适用于 InfluxDB 的 Timestream 中。包含 gzip 压缩行协议文件的目录作为命令行参数与摄取目标 InfluxDB 存储桶一起传递。该脚本采用多进程设计，可同时处理多个文件，从而充分利用 InfluxDB 及执行脚本的计算机资源。

脚本执行以下操作：
+ 提取压缩后的文件并将其摄取到 InfluxDB 中。
+ 实施重试机制和错误处理。
+ 跟踪成功和失败的摄取以便恢复。
+ 优化从线路协议数据集读取时的 I/O 操作。

**先决条件和安装**

请参阅摄取脚本的 [READ](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/README.md#installation) ME 中的 “先决条件和安装” 部分。 GitHub

**数据准备**

摄取所需的压缩行协议文件由数据转换脚本生成。按照以下步骤准备数据：

1. 设置具有充足存储空间的 EC2 实例，用于存放转换后的数据集。

1. 将转换后的数据从 S3 存储桶同步到本地目录：

   ```
   aws s3 sync \
       s3://your-bucket-name/path/to/transformed/data \
       ./data_directory
   ```

1. 确保您对数据目录中的所有文件都具有读取权限。

1. 运行以下摄取脚本，将数据摄取到适用于 InfluxDB 的 Timestream 中。

**用法**

```
python influxdb_ingestion.py <bucket_name> <data_directory> [options]
```

**基本用法**

```
python influxdb_ingestion.py my_bucket ./data_files
```

**摄取率**

我们已针对摄取率进行了一些测试。使用 C5N.9XL EC2 实例进行摄取测试，使用 10 个工作人员执行摄取脚本，并将大约 500 GB 的线路协议提取到 8XL Timestream 中用于 InfluxDB 实例：
+ 3K IOPS 15.86。 GB/hour
+ 12K IOPS 70.34。 GB/hour
+ 16K IOPS 71.28。 GB/hour

**建议**
+ 使用具有足够 CPU 内核的 EC2 实例进行并行处理。
+ 确保实例有足够的存储空间可容纳整个转换后的数据集，并预留额外空间用于提取操作。
  + 每次提取的文件数量等于脚本执行期间配置的工作线程数。
+ 将 EC2 实例部署在与 InfluxDB 实例相同的区域和可用区（如有可能），以最大限度地减少延迟。
+ 考虑使用针对网络操作进行优化的实例类型，例如 C5N。
+ 如果需要满足高吞吐量需求，建议为适用于 InfluxDB 的 Timestream 实例配置至少 12K IOPS。通过增加依赖适用于 InfluxDB 的 Timestream 实例大小的脚本的工作线程数量，可实现额外的优化效果。

有关更多信息，请参阅提取脚本的[自述文件](https://github.com/awslabs/amazon-timestream-tools/blob/mainline/tools/python/liveanalytics_migration_scripts/README.md#installation)。