AWS Data Pipeline 不再向新客户提供。的现有客户 AWS Data Pipeline 可以继续照常使用该服务。[了解详情](https://aws.amazon.com/blogs/big-data/migrate-workloads-from-aws-data-pipeline/)

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

# Expressions
<a name="dp-pipeline-expressions"></a>

利用表达式，您可以跨相关对象共享值。 AWS Data Pipeline Web 服务在运行时处理表达式，确保所有表达式都替换为表达式的值。

表达式用“\#{”和“}”分隔。您可以在字符串合法的任何管道定义对象中使用表达式。如果槽是一个引用，或者其类型为 ID、NAME、TYPE 或 SPHERE 之一，则不计算其值并且将按字面使用它。

以下表达式调用其中一个 AWS Data Pipeline 函数。有关更多信息，请参阅 [表达式计算](#dp-datatype-functions)。

```
#{format(myDateTime,'YYYY-MM-dd hh:mm:ss')}
```

## 引用字段和对象
<a name="dp-pipeline-expressions-reference"></a>

表达式既可以使用其所在的当前对象的字段，也可以使用通过引用链接的另一个对象的字段。

槽格式由创建时间后跟对象创建时间组成，例如 `@S3BackupLocation_2018-01-31T11:05:33`。

 您也可以引用管道定义中指定的确切的槽 ID，例如 Amazon S3 备份位置的槽 ID。要引用槽 ID，请使用 `#{parent.@id}`。

在以下示例中，`filePath` 字段引用同一对象中的 `id` 字段来生成文件名。`filePath` 值的计算结果为“`s3://amzn-s3-demo-bucket/ExampleDataNode.csv`”。

```
{
  "id" : "ExampleDataNode",
  "type" : "S3DataNode",
  "schedule" : {"ref" : "ExampleSchedule"},
  "filePath" : "s3://amzn-s3-demo-bucket/#{parent.@id}.csv",
  "precondition" : {"ref" : "ExampleCondition"},
  "onFail" : {"ref" : "FailureNotify"}
}
```

要使用通过引用链接的另一个对象上存在的字段，请使用 `node` 关键字。此关键字仅适用于警报和先决条件对象。

继续前一个示例，`SnsAlarm` 中的表达式可引用 `Schedule` 中的日期和时间范围，因为 `S3DataNode` 将引用二者。

 具体而言，`FailureNotify` 的 `message` 字段可从 `ExampleSchedule` 中使用 `@scheduledStartTime` 和 `@scheduledEndTime` 运行时字段，因为 `ExampleDataNode` 的 `onFail` 字段引用 `FailureNotify`，其 `schedule` 字段引用 `ExampleSchedule`。

```
{  
    "id" : "FailureNotify",
    "type" : "SnsAlarm",
    "subject" : "Failed to run pipeline component",
    "message": "Error for interval #{node.@scheduledStartTime}..#{node.@scheduledEndTime}.",
    "topicArn":"arn:aws:sns:us-east-1:28619EXAMPLE:ExampleTopic"
},
```

**注意**  
您可以创建具有依赖项的管道，例如，管道中的任务，这些任务依赖于其他系统或任务的工作。如果管道需要某些资源，请使用与数据节点和任务关联的先决条件将这些依赖项添加到管道。这可使管道更易于调试且更具弹性。此外，由于难以跨管道排查问题，因此，请将依赖项保留在单个管道中 (如果可能)。

## 嵌套表达式
<a name="dp-datatype-nested"></a>

 AWS Data Pipeline 允许您嵌套值以创建更复杂的表达式。例如，要执行时间计算 (从 `scheduledStartTime` 中减去 30 分钟) 并设置结果的格式以在管道定义中使用，您可以在活动中使用以下表达式：

```
#{format(minusMinutes(@scheduledStartTime,30),'YYYY-MM-dd hh:mm:ss')}
```

 如果表达式是 SnsAlarm 或前提条件的一部分，则使用前`node`缀：

```
#{format(minusMinutes(node.@scheduledStartTime,30),'YYYY-MM-dd hh:mm:ss')}
```

## Lists
<a name="dp-datatype-list-function"></a>

可基于列表和列表中的函数计算表达式。例如，假定按如下所示定义列表：`"myList":["one","two"]`。如果在表达式 `#{'this is ' + myList}` 中使用此列表，则该表达式的计算结果将为 `["this is one", "this is two"]`。如果您有两个列表，Data Pipeline 最终将在计算中平展列表。例如，如果 `myList1` 定义为 `[1,2]` 且 `myList2` 定义为 `[3,4]`，则表达式 `[#{myList1}, #{myList2}]` 的计算结果将为 `[1,2,3,4]`。

## 节点表达式
<a name="dp-datatype-node"></a>

 AWS Data Pipeline 使用`SnsAlarm`或中的`#{node.*}`表达式作为`PreCondition`对管道组件父对象的反向引用。由于 `SnsAlarm` 和 `PreCondition` 引用自一个活动或资源且未从其向后引用，因此，`node` 提供了对引用站点的引用方式。例如，以下管道定义演示故障通知如何使用 `node` 来引用其父级 (在此示例中，为 `ShellCommandActivity`)，并在 `SnsAlarm` 消息中包含父级的计划的开始和结束时间。上的 scheduledStartTime 引用 ShellCommandActivity 不需要`node`前缀，因为 scheduledStartTime 引用的是自身。

**注意**  
字段前面的 @ 符号指示这些字段是运行时字段。

```
{
  "id" : "ShellOut",
  "type" : "ShellCommandActivity",
  "input" : {"ref" : "HourlyData"},
  "command" : "/home/userName/xxx.sh #{@scheduledStartTime} #{@scheduledEndTime}",   
  "schedule" : {"ref" : "HourlyPeriod"},
  "stderr" : "/tmp/stderr:#{@scheduledStartTime}",
  "stdout" : "/tmp/stdout:#{@scheduledStartTime}",
  "onFail" : {"ref" : "FailureNotify"},
},
{  
  "id" : "FailureNotify",
  "type" : "SnsAlarm",
  "subject" : "Failed to run pipeline component",
  "message": "Error for interval #{node.@scheduledStartTime}..#{node.@scheduledEndTime}.",
  "topicArn":"arn:aws:sns:us-east-1:28619EXAMPLE:ExampleTopic"
},
```

AWS Data Pipeline 支持用户定义字段的传递引用，但不支持运行时字段。可传递引用是两个管道组件之间的引用，该引用依靠另一个管道组件作为中介。以下示例显示了一个对可传递的用户定义的字段的引用和一个对不可传递的运行时字段的引用，二者均有效。有关更多信息，请参阅 [用户定义字段](dp-writing-pipeline-definition.md#dp-userdefined-fields)。

```
{
  "name": "DefaultActivity1",
  "type": "CopyActivity",
  "schedule": {"ref": "Once"},
  "input": {"ref": "s3nodeOne"},  
  "onSuccess": {"ref": "action"},
  "workerGroup": "test",  
  "output": {"ref": "s3nodeTwo"}
},
{
  "name": "action",
  "type": "SnsAlarm",
  "message": "S3 bucket '#{node.output.directoryPath}' succeeded at #{node.@actualEndTime}.",
  "subject": "Testing",  
  "topicArn": "arn:aws:sns:us-east-1:28619EXAMPLE:ExampleTopic",
  "role": "DataPipelineDefaultRole"
}
```

## 表达式计算
<a name="dp-datatype-functions"></a>

 AWS Data Pipeline 提供了一组可用于计算字段值的函数。以下示例使用 `makeDate` 函数将 `Schedule` 对象的 `startDateTime` 字段设置为 `"2011-05-24T0:00:00"` GMT/UTC。

```
"startDateTime" : "makeDate(2011,5,24)"
```