# @aws-cdk/aws-pipes-alpha 中文文档

规范地址：https://npmcn.com/package/%40aws-cdk/aws-pipes-alpha/

版本：2\.275\.0-alpha\.0
npm 元数据更新时间：2026-10-26T19:19:43.029Z
中文译文更新时间：2026-11-07T13:54:39.972Z
软件包许可：Apache-2\.0

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40aws-cdk/aws-pipes-alpha
原文对照：https://npmcn.com/package/%40aws-cdk/aws-pipes-alpha/?lang=original
项目主页：https://github.com/aws/aws-cdk
源码仓库：https://github.com/aws/aws-cdk

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：602 kB，38 个文件
- 模块格式：CommonJS
- TypeScript 类型：包内自带
- Node.js 要求：\>= 20\.0\.0
- 运行依赖：0 个
- peer 依赖：constructs ^10\.5\.0，aws-cdk-lib ^2\.275\.0
- 首次发布：2024-02-01T02:45:05.830Z
- 最新版本发布：2026-10-26T19:19:42.067Z
- 近 30 天下载量：47903（2026-10-08 至 2026-11-06，npm 官方下载统计）
- 已发布版本数：199（近 12 个月 64 个）

## 中文简介

用于 Amazon EventBridge 管道的 CDK 构建库

## 安装

```sh
npm install @aws-cdk/aws-pipes-alpha
```

也可以使用：`pnpm add @aws-cdk/aws-pipes-alpha`、`yarn add @aws-cdk/aws-pipes-alpha`、`bun add @aws-cdk/aws-pipes-alpha`

## README

# 亚马逊事件 Bridge 管道构造库

* * *

![cdk-constructs: Experimental](https://img.shields.io/badge/cdk--constructs-experimental-important.svg?style=for-the-badge)

> 本单元中更高层次的构件的API是实验性的并正在积极开发中. 在今后任何版本中,它们都须作非背向相容的修改或去除。 这些是 不适用 [语义版本](https://semver.org/) 模型和断开的更改 在发行笔记中宣布. 这意味着虽然你可能使用它们,但可能需要更新 当更新到此软件包的更新版本时,您的源代码.

* * *

EventBridge Pipes 允许您创建源以瞄准多个连接 AWS服务. 当将消息从来源传送到目标时 可以过滤、转化和丰富.

![diagram of pipes](https://d1.awsstatic.com/product-marketing/EventBridge/Product-Page-Diagram_Amazon-EventBridge-Pipes.cd7961854be4432d63f6158ffd18271d6c9fa3ec.png)

详情见 [服务文件](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html)。 。 。 .

## 管道

[事件桥管](https://aws.amazon.com/blogs/aws/new-create-point-to-point-integrations-between-event-producers-and-consumers-with-amazon-eventbridge-pipes/) 是一种完全管理的服务,能够实现点对点之间的整合 事件生产者和消费者。 管道可用于连接多个 AWS 服务 互通,或连接AWS服务与外部服务.

管子有源和靶子. 源事件可以过滤并丰富 在到达目标之前.

## 示例 - 管道使用量

> 以下代码示例使用实例执行 a [来源](https://npmcn.com/package/%40aws-cdk/aws-pipes-alpha/#readme-source) 财务报告和审定财务报表 [目标](https://npmcn.com/package/%40aws-cdk/aws-pipes-alpha/#readme-target)。 。 。 .

要定义管道, 您需要创建一个新管道 `Pipe` 构造。 这个 `Pipe` 构造需要一个源和一个目标.

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue),
});
```

这个最小的例子创建一个以 SQS 队列为源而以 SQS 队列为目标的管道. 从源消息被放入目标消息正文.

## 来源

来源是接受调查的AWS服务. 可能的来源如下:

-   [亚马逊迪纳摩DB流](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-dynamodb.html)
-   [亚马逊克尼斯流](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-kinesis.html)
-   [亚马逊 MQ 经纪人](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-mq.html)
-   [亚马逊 MSK 流](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-msk.html)
-   [亚马逊 SQS 队列](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-sqs.html)
-   [阿帕奇·卡夫卡流](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-kafka.html)

目前,DynamoDB,Kinesis,以及SQS都得到了支持. 如果您有兴趣支持额外的来源, 请让我们知道,打开一个 GitHub 问题或提出公关.

### 示例来源

```ts
declare const sourceQueue: sqs.Queue;
const pipeSource = new SqsSource(sourceQueue);
```

## 过滤器

在事件发生之前, 可以用过滤器从源头过滤事件 被转移至浓缩物,如果没有浓缩物,则采取目标步骤。 多个过滤表达式是可能的. 如果过滤表达式之一匹配,事件会被传送到浓缩或目标步骤.

### 示例 - 过滤器使用

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;

const sourceFilter = new pipes.Filter(
  [
    pipes.FilterPattern.fromObject({
      body: {
        // only forward events with customerType B2B or B2C
        customerType: ['B2B', 'B2C'] 
      },
    })
  ]
)

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue),
  filter: sourceFilter, 
});
```

此示例显示一个过滤器,该过滤器仅转发与 `customerType` B2B或从源消息取出B2C. 不匹配过滤器的信件不会被传送到浓缩或目标步骤.

您可以定义多个过滤模式,这些模式与逻辑相结合 `OR`。 。 。 .

其他过滤图案和细节可见于 EventBridge 管道 [文档](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-filtering.html)。 。 。 .

## 输入转换

对于浓缩物和靶子,输入事件可以被改变. 转换适用于批次的每个项目. 变换可以访问输入事件,也可以访问管子本身的一些上下文信息,如管子的名称. 见 [文档](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-input-transformation.html) 详细情况.

### 示例 - 从对象输入转换

输入转换可以由对象产生. 对象可以包含静态值,动态值或管变量.

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;

const targetInputTransformation = pipes.InputTransformation.fromObject({
  staticField: 'static value',
  dynamicField: pipes.DynamicInput.fromEventPath('$.body.payload'),
  pipeVariable: pipes.DynamicInput.pipeName,
});

const pipe = new pipes.Pipe(this, 'Pipe', {
  pipeName: 'MyPipe',
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue, {
    inputTransformation: targetInputTransformation, 
  }),
});
```

这个例子显示了一个向输入事件添加静态字段,动态字段和管变量的转换. 动态字段取自输入事件. 管子变量取自管子上下文.

所以,当下面一批输入事件被管子处理

```json
[
  {
    ...
    "body": "{\"payload\": \"Test message.\"}",
    ...
  }
]
```

改为下列有效载荷:

```json
[
  {
    ...
    "staticField": "static value",
    "dynamicField": "Test message.",
    "pipeVariable": "MyPipe",
    ...
  }
]
```

如果将转换应用到目标上,它可能会被转换成字符串表示. 例如,由此产生的SQS消息正体看起来像这样:

```json
[
  {
    ...
    "body": "{\"staticField\": \"static value\", \"dynamicField\": \"Test message.\", \"pipeVariable\": \"MyPipe\"}",
    ...
  }
]
```

### 示例 - 从事件路径输入转换

如果您想要将事件的一部分转发给目标, 您可以使用转换事件路径 .

> 这只对目标有效,因为浓缩需要有一个有效的json作为输入.

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;

const targetInputTransformation = pipes.InputTransformation.fromEventPath('$.body.payload')

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue, {
    inputTransformation: targetInputTransformation, 
  }),
});
```

这种转变取出事体.

所以,当下面一批输入事件被管子处理

```json
 [
  {
    ...
    "body": "\"{\"payload\": \"Test message.\"}\"",
    ...
  }
]
```

它被转化为以下目标有效载荷:

```json
[
  {
    ...
    "body": "Test message."
    ...
  }
]
```

> 这个 [隐含的有效载荷分析](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-input-transformation.html#input-transform-implicit) (例如SQS消息正文给JSON)只有在输入是源有效载荷的情况下才有效. 隐形体剖析不适用于浓缩结果.

### 示例 - 从文本输入转换

如果您想要将静态文本转发到目标或使用自己的格式 `inputTemplate` 您可以使用文本中的转换.

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;

const targetInputTransformation = pipes.InputTransformation.fromText('My static text');

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue, {
    inputTransformation: targetInputTransformation, 
  }),
});
```

这种转变将静态文本推进到目标上.

```json
[
  {
    ...
    "body": "My static text"
    ...
  }
]
```

## 浓缩

在浓缩步骤中,来自源头的(未过滤出)有效载荷可用于援引下列一种服务:

-   API 目的地
-   亚马逊 API 网关
-   Lambda 函数
-   Step 函数状态机器
    -   只表达工作流程

### 实例浓缩实施

> 目前没有任何支持的浓缩措施得到执行。 以下例子说明执行的外观。 实际执行不是这一一揽子计划的一部分,而是单独执行.

```ts
class LambdaEnrichment implements pipes.IEnrichment {
  enrichmentArn: string;
  private inputTransformation: pipes.InputTransformation  | undefined;
  
  constructor(private readonly lambda: lambda.Function, props: {inputTransformation?: pipes.InputTransformation} = {}) {
    this.enrichmentArn = lambda.functionArn;
    this.inputTransformation = props?.inputTransformation
  }
  
  bind(pipe: pipes.IPipe): pipes.EnrichmentParametersConfig {
    return {
      enrichmentParameters: {
        inputTemplate: this.inputTransformation?.bind(pipe).inputTemplate,
      },
    };
  }
  
  grantInvoke(pipeRole: cdk.aws_iam.IRole): void {
    this.lambda.grantInvoke(pipeRole);
  }
}
```

浓缩实施需要提供 `enrichmentArn`, (中文(简体) ), `enrichmentParameters` 并赋予管道角色 援引获得浓缩.

### 示例 - 浓缩使用

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;
declare const enrichmentLambda: lambda.Function;

const enrichmentInputTransformation = pipes.InputTransformation.fromObject({
  staticField: 'static value',
  dynamicField: pipes.DynamicInput.fromEventPath('$.body.payload'),
  pipeVariable: pipes.DynamicInput.pipeName,
});

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue),
  enrichment: new LambdaEnrichment(enrichmentLambda, {
    inputTransformation: enrichmentInputTransformation,
  }),
});
```

这个例子增加了羊肉达功能作为管道的浓缩. lambda函数在应用了变换后随批发源信息一起被引用. lambda 函数可以返回向目标传输的结果.

因此,下面一批输入事件由管子处理

```json
[
  {
    ...
    "body": "{\"payload\": \"Test message.\"}",
    ...
  }
]
```

它被转换成以下有效载荷,并被传送到羊肉达函数上.

```json
[
  {
    ...
    "staticField": "static value",
    "dynamicField": "Test message.",
    "pipeVariable": "MyPipe",
    ...
  }
]
```

Lambda 函数可以返回向目标传输的结果. 例如, lambda 函数返回静态字段、 动态字段和管变量的连接

```ts
export async function handler (event: any) {
  return event.staticField + "-" + event.dynamicField + "-" + event.pipeVariable;
};
```

将在目标 SQS 队列中生成以下目标消息.

```json
[
  {
    ...
    "body": "static value-Test message.-MyPipe",
    ...
  }
]
```

## 目标

一个目标就是管道的尽头 在从源头取出有效载荷后, 过滤和浓缩后转呈目标。 目前情况如下: 目标得到支持:

-   [API 目的地](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-api-destinations.html)
-   [API 网关](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-api-gateway-target.html)
-   [批量任务队列](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-target.html#pipes-targets-specifics-batch)
-   [云表日志组](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-target.html#pipes-targets-specifics-cwl)
-   [ECS任务](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-event-target.html#pipes-targets-specifics-ecs-task)
-   同一账户和区域的活动总线
-   火花快递流
-   检查员评估模板
-   Kinesis 溪流
-   Lambda 函数(SYNC或ASYNC)
-   红移集群数据 API 查询
-   SageMaker 管道
-   SNS专题
-   SQS 队列
-   Step 函数状态机器
    -   快速工作流程(ASYNC)
    -   标准工作流程(SYNC或ASYNC)

目标事件在传送到目标之前可以转换,使用 与浓缩步骤相同的输入转换.

### 示例目标

```ts
declare const targetQueue: sqs.Queue;
const pipeTarget = new SqsTarget(targetQueue);
```

## 日志目标

管可以产生日志事件,被传送到不同的日志目的地. 您可以配置多个目的地, 但所有目的地共享相同的日志级别和日志数据 . 详细查一下官员 [文档](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes-logs.html)。 。 。 .

日志事件包含的日志级别和数据在管类本身上配置. 实际目的地是独立的,有三种选择:

1.  `CloudwatchLogsLogDestination`
2.  `FirehoseLogDestination`
3.  `S3LogDestination`

### 日志目的使用示例

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;
declare const logGroup: logs.LogGroup;

const cwlLogDestination = new pipes.CloudwatchLogsLogDestination(logGroup);

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue),
  logLevel: pipes.LogLevel.TRACE,
  logIncludeExecutionData: [pipes.IncludeExecutionData.ALL],
  logDestinations: [cwlLogDestination],
});
```

此示例使用 CloudWatch 日志组来存储管道执行过程中释放的日志 . 日志级别设定为 `TRACE` 所以管道的所有步骤都被记录下来了. 此外,还记录了所有执行数据.

## 用 KMS 加密管道数据

您可以指定 EventBridge 使用客户管理的密钥加密存储在休息时的管道数据, 而不是像默认那样使用 AWS 拥有的密钥. 详情见 [文档](https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-encryption-pipes-cmkey.html)。 。 。 .

要做到这一点,请在 `kmsKey` 管道的属性.

```ts
declare const sourceQueue: sqs.Queue;
declare const targetQueue: sqs.Queue;
declare const kmsKey: kms.Key;

const pipe = new pipes.Pipe(this, 'Pipe', {
  source: new SqsSource(sourceQueue),
  target: new SqsTarget(targetQueue),
  kmsKey,
  // pipeName is required when using a KMS key
  pipeName: 'MyPipe',
});
```

## 常见问题

### @aws-cdk/aws-pipes-alpha 是什么？

用于 Amazon EventBridge 管道的 CDK 构建库

### 如何安装 @aws-cdk/aws-pipes-alpha？

在项目目录运行 npm install @aws-cdk/aws-pipes-alpha，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：constructs（^10\.5\.0）、aws-cdk-lib（^2\.275\.0）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

### @aws-cdk/aws-pipes-alpha 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @aws-cdk/aws-pipes-alpha 支持 ESM 还是 CommonJS？

以 CommonJS 形式发布，可以直接 require\(\)；在 ES Module 项目中也可以用 import 引入（Node\.js 会把 module\.exports 作为默认导出）。

### @aws-cdk/aws-pipes-alpha 需要什么版本的 Node\.js？

package\.json 的 engines 字段要求 Node\.js \>= 20\.0\.0。版本不符时 npm 默认只给出警告，开启 engine-strict 后会拒绝安装。

### @aws-cdk/aws-pipes-alpha 可以商用吗？

@aws-cdk/aws-pipes-alpha 使用 Apache-2\.0 许可。宽松许可：允许商业使用、修改和再分发，并附带明确的专利授权；分发时需保留许可与 NOTICE 文件，修改过的文件要注明改动。以上为概括，不构成法律意见，请以许可原文为准。

### @aws-cdk/aws-pipes-alpha 的下载量有多少？

根据 npm 官方下载统计，@aws-cdk/aws-pipes-alpha 在 2026/10/08 – 2026/11/06 这 30 天内被下载 47,903 次，日均约 1,597 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @aws-cdk/aws-pipes-alpha 还在维护吗？

最新版本 2\.275\.0-alpha\.0 发布于 2026/10/27（12 天前）；至今共发布 199 个版本，其中过去 12 个月发布了 64 个；首次发布于 2024/02/01。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。