本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 為 Amazon MSK 事件來源建立 Lambda 事件來源映射
<a name="msk-esm-create"></a>

若要建立事件來源映射，可以使用 Lambda 主控台、[AWS Command Line Interface (CLI)](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) 或 [AWS SDK](https://aws.amazon.com/getting-started/tools-sdks/)。

**注意**  
當您建立事件來源映射時，Lambda 會在包含 MSK 叢集的私有子網路內建立 [Hyperplane ENI](configuration-vpc.md#configuration-vpc-enis)，使 Lambda 能建立安全連線。此 Hyperplane ENI 會使用 MSK 叢集的子網路和安全群組組態，而不是 Lambda 函式。

下列主控台步驟會將 Amazon MSK 叢集新增為 Lambda 函式的觸發程序。在此之下，此操作會建立事件來源映射資源。

**將 Amazon MSK 觸發條件新增至 Lambda 函數 （主控台）**

1. 開啟 Lambda 主控台中的[函數頁面](https://console.aws.amazon.com/lambda/home#/functions)。

1. 選擇要新增 Amazon MSK 觸發程序的 Lambda 函式名稱。

1. 在**函數概觀**下，選擇**新增觸發條件**。

1. 在**觸發程序組態**欄位中，選擇 **MSK**。

1. 若要指定 Kafka 叢集詳細資訊，請執行下列動作：

   1. 對於 **MSK cluster (MSK 叢集)**，請選取您的叢集。

   1. 在**主題名稱**欄位中，輸入要從中取用訊息的 Kafka 主題名稱。

   1. (選用) 在**取用者群組 ID** 欄位中，輸入要加入的 Kafka 取用者群組 ID (如適用)。如需詳細資訊，請參閱[Lambda 中可自訂的取用者群組 ID](kafka-consumer-group-id.md)。

1. 在**叢集身分驗證**欄位中，進行必要的組態設定。如需有關叢集身分驗證的詳細資訊，請參閱[在 Lambda 中設定 Amazon MSK 叢集身分驗證方法](msk-cluster-auth.md)。
   + 如果希望 Lambda 在建立連線時對 MSK 叢集執行身分驗證，請開啟**使用身分驗證**。建議進行身分驗證。
   + 如果使用身分驗證，在**身分驗證方法**欄位中，選擇要使用的身分驗證方法。
   + 如果使用身分驗證，在 **Secrets Manager 金鑰**欄位中，選擇包含存取叢集所需身分驗證憑證的 Secrets Manager 金鑰。

1. 在**事件輪詢器組態**欄位中，進行必要的組態設定。
   + 選擇**啟用觸發程序**，在建立完成後立即啟用觸發程序。
   + 選擇是否要為事件來源映射**設定佈建模式**。如需詳細資訊，請參閱[Lambda 中的 Apache Kafka 事件輪詢器擴展模式](kafka-scaling-modes.md)。
     + 如果您設定佈建模式，請輸入**最小值事件輪詢器**的值、**最大值事件輪詢器**的值，以及 PollerGroupName 的選用值，以指定相同事件來源 VPC 內多個 ESMs 的分組。
   + 在**開始位置**欄位中，選擇希望 Lambda 開始從串流讀取的方式。如需詳細資訊，請參閱[Lambda 中的 Apache Kafka 輪詢與串流開始位置](kafka-starting-positions.md)。

1. 在**批次處理**欄位中，進行必要的組態設定。如需有關批次處理的詳細資訊，請參閱[批次處理行為](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching)。

   1. 對於 **批次大小**，輸入單一批次中要擷取的訊息數量上限。

   1. 在**批次時段**欄位中，輸入 Lambda 調用函式之前收集記錄所耗費的秒數上限。

1. 在**篩選**欄位中，進行必要的組態設定。如需有關篩選的詳細資訊，請參閱 [從 Amazon MSK 和自我管理的 Apache Kafka 事件來源篩選事件](kafka-filtering.md)。
   + 在**篩選條件**欄位中，新增篩選條件定義，決定是否處理事件。

1. 在**失敗處理**欄位中，進行必要的組態設定。如需有關失敗處理的詳細資訊，請參閱[擷取 Amazon MSK 和自我管理的 Apache Kafka 事件來源的捨棄批次](kafka-on-failure.md)。
   + 在**失敗時目的地**欄位中，指定失敗時目的地的 ARN。

1. 在**標籤**欄位中，輸入要與此事件來源映射建立關聯的標籤。

1. 若要建立觸發條件，請選擇 **新增** 。

您也可以使用 CLI AWS 搭配 [ create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) 命令來建立事件來源映射。下列範例建立了事件來源映射，可將 Lambda 函式 `my-msk-function` 映射至 `AWSKafkaTopic` 主題 (從 `LATEST` 訊息開始)。此命令還使用了 [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) 物件，可指示 Lambda 在連線至叢集時使用 [SASL/SCRAM](msk-cluster-auth.md#msk-sasl-scram) 身分驗證。

```
aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
```

如果叢集使用 [mTLS 身分驗證](msk-cluster-auth.md#msk-mtls)，則需包含指定 `CLIENT_CERTIFICATE_TLS_AUTH` 和 Secrets Manager 金鑰 ARN 的 [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) 物件。如下列命令所示：

```
aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
```

如果叢集使用 [IAM 身分驗證](msk-cluster-auth.md#msk-iam-auth)，則不需要 [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) 物件。如下列命令所示：

```
aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
```