AWS Data Pipeline 不再向新客户提供。的现有客户 AWS Data Pipeline 可以继续照常使用该服务。[了解详情](https://aws.amazon.com/blogs/big-data/migrate-workloads-from-aws-data-pipeline/)
本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 数据格式
以下是 AWS Data Pipeline 数据格式对象：
**Topics**
+ [CSV 数据格式](dp-object-csv.md)
+ [自定义数据格式](dp-object-custom.md)
+ [Dynamo 格式 DBData](dp-object-dynamodbdataformat.md)
+ [迪纳摩 DBExport DataFormat](dp-object-dynamodbexportdataformat.md)
+ [RegEx 数据格式](dp-object-regex.md)
+ [TSV 数据格式](dp-object-tsv.md)
# CSV 数据格式
一种逗号分隔的数据格式，其中列分隔符是逗号，记录分隔符是换行符。
## 示例
以下是该对象类型的示例。
```
{
"id" : "MyOutputDataType",
"type" : "CSV",
"column" : [
"Name STRING",
"Score INT",
"DateOfBirth TIMESTAMP"
]
}
```
## 语法
****
| 可选字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| column | 具有该数据节点描述的数据的各个字段指定的数据类型的列名。例如：hostname STRING。对于多个值，使用空格分隔的列名和数据类型。 | 字符串 |
| escapeChar | 一个字符，例如“\$1”，指示分析器忽略下一个字符。 | 字符串 |
| parent | 槽将继承自的当前对象的父级。 | 引用对象，例如 “父对象”：\$1"ref”:” myBaseObject Id "\$1 |
****
| 运行时字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @version | 用来创建对象的管道版本。 | 字符串 |
****
| 系统字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @error | 用于描述格式不正确的对象的错误消息 | 字符串 |
| @pipelineId | 该对象所属的管道的 ID | 字符串 |
| @sphere | 对象的范围指明对象在生命周期中的位置：组件对象产生实例对象，后者执行尝试对象 | 字符串 |
# 自定义数据格式
由特定的列分隔符、记录分隔符和转义字符组合定义的自定义数据格式。
## 示例
以下是该对象类型的示例。
```
{
"id" : "MyOutputDataType",
"type" : "Custom",
"columnSeparator" : ",",
"recordSeparator" : "\n",
"column" : [
"Name STRING",
"Score INT",
"DateOfBirth TIMESTAMP"
]
}
```
## 语法
****
| 必填字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| columnSeparator | 一个指示数据文件中的列结尾的字符。 | 字符串 |
****
| 可选字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| column | 具有该数据节点描述的数据的各个字段指定的数据类型的列名。例如：hostname STRING。对于多个值，使用空格分隔的列名和数据类型。 | 字符串 |
| parent | 槽将继承自的当前对象的父级。 | 引用对象，例如 “父对象”：\$1"ref”:” myBaseObject Id "\$1 |
| recordSeparator | 一个指示数据文件中的行结尾的字符，例如“\$1n”。仅支持单个字符。 | 字符串 |
****
| 运行时字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @version | 用来创建对象的管道版本。 | 字符串 |
****
| 系统字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @error | 用于描述格式不正确的对象的错误消息 | 字符串 |
| @pipelineId | 该对象所属的管道的 ID | 字符串 |
| @sphere | 对象的范围指明对象在生命周期中的位置：组件对象产生实例对象，后者执行尝试对象 | 字符串 |
# Dynamo 格式 DBData
将架构应用于 DynamoDB 表以使其可供 Hive 查询访问。`DynamoDBDataFormat` 与 `HiveActivity` 对象以及 `DynamoDBDataNode` 输入和输出一起使用。`DynamoDBDataFormat` 要求您在 Hive 查询中指定所有列。要更灵活地在 Hive 查询或 Amazon S3 支持中指定特定列，请参阅 [迪纳摩 DBExport DataFormat](dp-object-dynamodbexportdataformat.md)。
**注意**
DynamoDB 布尔类型不会映射到 Hive 布尔类型。不过，可以将 DynamoDB 整数值 0 或 1 映射到 Hive 布尔类型。
## 示例
以下示例说明如何使用 `DynamoDBDataFormat` 将架构分配给 `DynamoDBDataNode` 输入，这将允许 `HiveActivity` 对象按命名的列访问数据并将数据复制到 `DynamoDBDataNode` 输出。
```
{
"objects": [
{
"id" : "Exists.1",
"name" : "Exists.1",
"type" : "Exists"
},
{
"id" : "DataFormat.1",
"name" : "DataFormat.1",
"type" : "DynamoDBDataFormat",
"column" : [
"hash STRING",
"range STRING"
]
},
{
"id" : "DynamoDBDataNode.1",
"name" : "DynamoDBDataNode.1",
"type" : "DynamoDBDataNode",
"tableName" : "$INPUT_TABLE_NAME",
"schedule" : { "ref" : "ResourcePeriod" },
"dataFormat" : { "ref" : "DataFormat.1" }
},
{
"id" : "DynamoDBDataNode.2",
"name" : "DynamoDBDataNode.2",
"type" : "DynamoDBDataNode",
"tableName" : "$OUTPUT_TABLE_NAME",
"schedule" : { "ref" : "ResourcePeriod" },
"dataFormat" : { "ref" : "DataFormat.1" }
},
{
"id" : "EmrCluster.1",
"name" : "EmrCluster.1",
"type" : "EmrCluster",
"schedule" : { "ref" : "ResourcePeriod" },
"masterInstanceType" : "m1.small",
"keyPair" : "$KEYPAIR"
},
{
"id" : "HiveActivity.1",
"name" : "HiveActivity.1",
"type" : "HiveActivity",
"input" : { "ref" : "DynamoDBDataNode.1" },
"output" : { "ref" : "DynamoDBDataNode.2" },
"schedule" : { "ref" : "ResourcePeriod" },
"runsOn" : { "ref" : "EmrCluster.1" },
"hiveScript" : "insert overwrite table ${output1} select * from ${input1} ;"
},
{
"id" : "ResourcePeriod",
"name" : "ResourcePeriod",
"type" : "Schedule",
"period" : "1 day",
"startDateTime" : "2012-05-04T00:00:00",
"endDateTime" : "2012-05-05T00:00:00"
}
]
}
```
## 语法
****
| 可选字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| column | 具有该数据节点描述的数据的各个字段指定的数据类型的列名。例如 hostname STRING。对于多个值，请使用以空格分隔的列名和数据类型。 | 字符串 |
| parent | 槽将继承自的当前对象的父级。 | 引用对象，例如 “父对象”：\$1"ref”:” myBaseObject Id "\$1 |
****
| 运行时字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @version | 用于创建对象的管道版本。 | 字符串 |
****
| 系统字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @error | 用于描述格式不正确的对象的错误消息。 | 字符串 |
| @pipelineId | 该对象所属的管道的 ID。 | 字符串 |
| @sphere | 对象的范围指明对象在生命周期中的位置：组件对象产生实例对象，后者执行尝试对象。 | 字符串 |
# 迪纳摩 DBExport DataFormat
将架构应用于 DynamoDB 表以使其可供 Hive 查询访问。将 `DynamoDBExportDataFormat` 与 `HiveCopyActivity` 对象和 `DynamoDBDataNode` 或 `S3DataNode` 输入和输出一起使用。`DynamoDBExportDataFormat` 具有以下优势：
+ 提供 DynamoDB 和 Amazon S3 支持
+ 允许您在 Hive 查询中按特定列筛选数据
+ 导出 DynamoDB 中的所有属性（即使您有一个稀疏架构）
**注意**
DynamoDB 布尔类型不会映射到 Hive 布尔类型。不过，可以将 DynamoDB 整数值 0 或 1 映射到 Hive 布尔类型。
## 示例
以下示例说明如何使用 `HiveCopyActivity` 和 `DynamoDBExportDataFormat` 将数据从一个 `DynamoDBDataNode` 复制到另一个 DynamoDBDataNode，并基于时间戳进行筛选。
```
{
"objects": [
{
"id" : "DataFormat.1",
"name" : "DataFormat.1",
"type" : "DynamoDBExportDataFormat",
"column" : "timeStamp BIGINT"
},
{
"id" : "DataFormat.2",
"name" : "DataFormat.2",
"type" : "DynamoDBExportDataFormat"
},
{
"id" : "DynamoDBDataNode.1",
"name" : "DynamoDBDataNode.1",
"type" : "DynamoDBDataNode",
"tableName" : "item_mapped_table_restore_temp",
"schedule" : { "ref" : "ResourcePeriod" },
"dataFormat" : { "ref" : "DataFormat.1" }
},
{
"id" : "DynamoDBDataNode.2",
"name" : "DynamoDBDataNode.2",
"type" : "DynamoDBDataNode",
"tableName" : "restore_table",
"region" : "us_west_1",
"schedule" : { "ref" : "ResourcePeriod" },
"dataFormat" : { "ref" : "DataFormat.2" }
},
{
"id" : "EmrCluster.1",
"name" : "EmrCluster.1",
"type" : "EmrCluster",
"schedule" : { "ref" : "ResourcePeriod" },
"masterInstanceType" : "m1.xlarge",
"coreInstanceCount" : "4"
},
{
"id" : "HiveTransform.1",
"name" : "Hive Copy Transform.1",
"type" : "HiveCopyActivity",
"input" : { "ref" : "DynamoDBDataNode.1" },
"output" : { "ref" : "DynamoDBDataNode.2" },
"schedule" : { "ref" : "ResourcePeriod" },
"runsOn" : { "ref" : "EmrCluster.1" },
"filterSql" : "`timeStamp` > unix_timestamp(\"#{@scheduledStartTime}\", \"yyyy-MM-dd'T'HH:mm:ss\")"
},
{
"id" : "ResourcePeriod",
"name" : "ResourcePeriod",
"type" : "Schedule",
"period" : "1 Hour",
"startDateTime" : "2013-06-04T00:00:00",
"endDateTime" : "2013-06-04T01:00:00"
}
]
}
```
## 语法
****
| 可选字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| column | 具有该数据节点描述的数据的各个字段指定的数据类型的列名。例如：hostname STRING | 字符串 |
| parent | 槽将继承自的当前对象的父级。 | 引用对象，例如 “父对象”：\$1"ref”:” myBaseObject Id "\$1 |
****
| 运行时字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @version | 用来创建对象的管道版本。 | 字符串 |
****
| 系统字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @error | 用于描述格式不正确的对象的错误消息 | 字符串 |
| @pipelineId | 该对象所属的管道的 ID | 字符串 |
| @sphere | 对象的范围指明对象在生命周期中的位置：组件对象产生实例对象，后者执行尝试对象 | 字符串 |
# RegEx 数据格式
由正则表达式定义的自定义数据格式。
## 示例
以下是该对象类型的示例。
```
{
"id" : "MyInputDataType",
"type" : "RegEx",
"inputRegEx" : "([^ ]*) ([^ ]*) ([^ ]*) (-|\\[[^\\]]*\\]) ([^ \"]*|\"[^\"]*\") (-|[0-9]*) (-|[0-9]*)(?: ([^ \"]*|\"[^\"]*\") ([^ \"]*|\"[^\"]*\"))?",
"outputFormat" : "%1$s %2$s %3$s %4$s %5$s %6$s %7$s %8$s %9$s",
"column" : [
"host STRING",
"identity STRING",
"user STRING",
"time STRING",
"request STRING",
"status STRING",
"size STRING",
"referer STRING",
"agent STRING"
]
}
```
## 语法
****
| 可选字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| column | 具有该数据节点描述的数据的各个字段指定的数据类型的列名。例如：hostname STRING。对于多个值，使用空格分隔的列名和数据类型。 | 字符串 |
| inputRegEx | 用于解析 S3 输入文件的正则表达式。 inputRegEx 提供了一种从文件中相对非结构化的数据中检索列的方法。 | 字符串 |
| outputFormat | 由 Java 格式化程序语法 inputRegEx检索但引用为 %1\$1s %2\$1s 的列字段。 | 字符串 |
| parent | 槽将继承自的当前对象的父级。 | 引用对象，例如 “父对象”：\$1"ref”:” myBaseObject Id "\$1 |
****
| 运行时字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @version | 用来创建对象的管道版本。 | 字符串 |
****
| 系统字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @error | 用于描述格式不正确的对象的错误消息 | 字符串 |
| @pipelineId | 该对象所属的管道的 ID | 字符串 |
| @sphere | 对象的范围指明对象在生命周期中的位置：组件对象产生实例对象，后者执行尝试对象 | 字符串 |
# TSV 数据格式
一种逗号分隔的数据格式，其中列分隔符是制表符，记录分隔符是换行符。
## 示例
以下是该对象类型的示例。
```
{
"id" : "MyOutputDataType",
"type" : "TSV",
"column" : [
"Name STRING",
"Score INT",
"DateOfBirth TIMESTAMP"
]
}
```
## 语法
****
| 可选字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| column | 该数据节点描述的数据的列名和数据类型。例如，"Name STRING" 表示名为 Name 的列，其字段的数据类型为 STRING。请使用逗号分隔多个列名和数据类型对 (如示例中所示)。 | 字符串 |
| columnSeparator | 将一列中的字段与下一列中字段分隔的字符。默认为“\$1t”。 | 字符串 |
| escapeChar | 一个字符，例如“\$1”，指示分析器忽略下一个字符。 | 字符串 |
| parent | 作为槽继承源的当前对象的父项。 | 参考对象，例如，“父对象”：\$1"ref”:” myBaseObject Id "\$1 |
| recordSeparator | 分隔记录的字符。默认为“\$1n”。 | 字符串 |
****
| 运行时字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @version | 用于创建对象的管道版本。 | 字符串 |
****
| 系统字段 | 说明 | 槽位类型 |
| --- | --- | --- |
| @error | 用于描述格式不正确的对象的错误消息。 | 字符串 |
| @pipelineId | 该对象所属的管道的 ID。 | 字符串 |
| @sphere | 对象的范围指明对象在生命周期中的位置：组件对象产生实例对象，后者执行尝试对象。 | 字符串 |