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

# 使用 Lambda 處理來自 Amazon Kinesis Data Streams 的記錄
<a name="with-kinesis"></a>

您可以使用 Lambda 事件函數來處理 [Amazon Kinesis 資料串流](https://docs.aws.amazon.com/streams/latest/dev/introduction.html)中的記錄。您可以將 Lambda 事件函數映射至 Kinesis Data Streams 共用輸送量取用者 （標準迭代器），或映射至具有[增強廣發](https://docs.aws.amazon.com/kinesis/latest/dev/enhanced-consumers.html)功能的專用輸送量取用者。對於標準迭代程式，Lambda 會使用 HTTP 協定輪詢 Kinesis 串流中的每個碎片以尋找記錄。事件來源映射會與碎片的其他取用者共用讀取傳輸量。

 如需 Kinesis 資料串流的詳細資訊，請參閱[從 Amazon Kinesis Data Streams 讀取資料](https://docs.aws.amazon.com/kinesis/latest/dev/building-consumers.html)。

若要開始使用，請參閱[使用 Lambda 處理 Amazon Kinesis Data Streams 記錄](services-kinesis-create.md)。您可以設定[事件來源映射參數](services-kinesis-parameters.md)、使用[輪轉時段](services-kinesis-windows.md)進行彙總，以及檢視[範例函數程式碼](with-kinesis-example.md)。

**注意**  
Kinesis 會收取每個碎片的費用，以及從串流讀取資料的增強型散發。如需定價的詳細資訊，請參閱 [Amazon Kinesis 定價](https://aws.amazon.com/kinesis/data-streams/pricing)。

## 輪詢和批次處理串流
<a name="kinesis-polling-and-batching"></a>

Lambda 會從資料串流讀取記錄並透過包含串流記錄的事件[同步](invocation-sync.md)調用函數。Lambda 會讀取批次中的記錄並調用函數，以處理來自該批次的記錄。每個批次包含來自單一碎片/資料串流的記錄。

您的 Lambda 事件函數是資料串流的取用者應用程式。其會從每個碎片中一次處理一個批次的記錄。您可以將 Lambda 事件函數映射至共用輸送量消費者 （標準迭代器），或映射至具有增強廣發功能的專用輸送量消費者。
+ **標準迭代器：**Lambda 會輪詢 Kinesis 串流中的每個碎片，其記錄的基本速率為每秒一次。有更多記錄可用時，Lambda 會持續處理批次，直到函數掌握串流資訊。事件來源映射會與碎片的其他取用者共用讀取傳輸量。
+ **增強散發功能：**若要將延遲降至最低，並將讀取傳輸量增至最大，請建立具有[增強散發功能](https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html)的資料串流取用者。增強散發取用者會取得每個碎片的專用連線，其不會影響其他從串流讀取的應用程式。串流取用者會使用 HTTP/2 來減少延遲，方法為透過長期連線將記錄推送至 Lambda，以及透過壓縮請求標頭。您可以使用 Kinesis [RegisterStreamConsumer](https://docs.aws.amazon.com/kinesis/latest/APIReference/API_RegisterStreamConsumer.html) API，建立串流取用者。

```
aws kinesis register-stream-consumer \
--consumer-name con1 \
--stream-arn arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream
```

您應該會看到下列輸出：

```
{
    "Consumer": {
        "ConsumerName": "con1",
        "ConsumerARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream/consumer/con1:1540591608",
        "ConsumerStatus": "CREATING",
        "ConsumerCreationTimestamp": 1540591608.0
    }
}
```

若要增加函數處理記錄的速度，[請將碎片新增至您的資料串流](https://repost.aws/knowledge-center/kinesis-data-streams-open-shards)。Lambda 會依序在每個碎片中處理記錄。如果您的函數傳回錯誤，則會停止處理碎片中的其他記錄。碎片越多，要一次處理的批次越多，這會降低錯誤對並行的影響。

如果您的函數無法擴展至可處理並行批次的總數，您可以[請求增加配額](https://docs.aws.amazon.com/servicequotas/latest/userguide/request-quota-increase.html)，或為您的函數[保留並行](configuration-concurrency.md)。

Lambda 預設會在記錄可用時立即調用函數。如果 Lambda 從事件來源中讀取的批次只有一筆記錄，Lambda 只會傳送一筆記錄至函數。為避免調用具有少量記錄的函數，您可設定*批次間隔*，請求事件來源緩衝記錄最長達五分鐘。調用函數之前，Lambda 會繼續從事件來源中讀取記錄，直到收集到完整批次、批次間隔到期或者批次達到 6 MB 的承載限制。如需詳細資訊，請參閱[批次處理行為](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching)。

**警告**  
Lambda 事件來源映射至少會處理每個事件一次，而且可能會重複處理記錄。為避免與重複事件相關的潛在問題，強烈建議您讓函數程式碼具有等冪性。若要進一步了解，請參閱 AWS 知識中心中的[如何使 Lambda 函數具有等冪性](https://repost.aws/knowledge-center/lambda-function-idempotent)。

Lambda 在傳送下一批進行處理前不會等待任何已設定的[擴充](lambda-extensions.md)完成。換句話說，當 Lambda 處理下一批記錄時，您的擴充功能可能會繼續執行。如果您違反任何帳戶的 [並行](lambda-concurrency.md) 設定或限制，便可能會產生限流的問題。若要偵測此是否為潛在問題，請監控您的函數，並確認您看到的 [並行指標](monitoring-concurrency.md#general-concurrency-metrics) 是否高於事件來源映射的預期值。由於兩次調用之間的時間很短，Lambda 可能會短暫報告比碎片數目更高的並行用量。即使 Lambda 事件函數沒有擴充功能，也是如此。

設定 [ParallelizationFactor](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html#lambda-CreateEventSourceMapping-request-ParallelizationFactor) 設定，同時使用多個 Lambda 調用來處理 Kinesis 資料串流的一個碎片。您可以使用從 1 （預設） 到 10 的平行化係數，指定 Lambda 從碎片輪詢的並行批次數。例如，當您將 `ParallelizationFactor`設定為 2 時，最多可以有 200 個並行 Lambda 調用來處理 100 個 Kinesis 資料碎片 （實際上，您可能會看到`ConcurrentExecutions`指標的不同值）。當資料量急劇波動並且 `IteratorAge` 高時，這有助於縱向擴展處理輸送量。如果增加每個碎片的並行批次數量，則 Lambda 仍會確保在分割區索引鍵層級進行順序處理。

您也可以搭配 Kinesis 彙總使用 `ParallelizationFactor`。事件來源映射的行為取決於您是否使用[增強型散發](https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html)：
+ **沒有使用增強型散發**：彙總事件內的所有事件都必須具有相同的分割區索引鍵。分割區索引鍵也必須與彙總事件的分割區索引鍵相符。如果彙總事件內的事件具有不同的分割區索引鍵，則 Lambda 無法保證依分割區索引鍵依序處理事件。
+ **使用增強型散發**：首先，Lambda 會將彙總的事件解碼為其個別事件。彙總的事件可以具有與其包含的事件不同的分割區索引鍵。不過，未對應至分割區索引鍵的事件會[遭到捨棄和遺失](https://github.com/awslabs/kinesis-aggregation/blob/master/potential_data_loss.md)。Lambda 不會處理這些事件，也不會將其傳送至設定的失敗目的地。

## 範例事件
<a name="services-kinesis-event-example"></a>

**Example**  

```
{
    "Records": [
        {
            "kinesis": {
                "kinesisSchemaVersion": "1.0",
                "partitionKey": "1",
                "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
                "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
                "approximateArrivalTimestamp": 1545084650.987
            },
            "eventSource": "aws:kinesis",
            "eventVersion": "1.0",
            "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
            "eventName": "aws:kinesis:record",
            "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
            "awsRegion": "us-east-2",
            "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
        },
        {
            "kinesis": {
                "kinesisSchemaVersion": "1.0",
                "partitionKey": "1",
                "sequenceNumber": "49590338271490256608559692540925702759324208523137515618",
                "data": "VGhpcyBpcyBvbmx5IGEgdGVzdC4=",
                "approximateArrivalTimestamp": 1545084711.166
            },
            "eventSource": "aws:kinesis",
            "eventVersion": "1.0",
            "eventID": "shardId-000000000006:49590338271490256608559692540925702759324208523137515618",
            "eventName": "aws:kinesis:record",
            "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
            "awsRegion": "us-east-2",
            "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
        }
    ]
}
```