您首先需要構(gòu)建 Docker 鏡像并將其推送到Amazon Elastic Container Registry (ECR)（有關(guān)Dockerfile詳細(xì)信息，請(qǐng)參閱 ）。

# create runnable JAR filemvn clean compile assembly\:single# build docker imagedocker build -t kcl-consumer-app .AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query "Account" --output text)# create a private ECR repoaws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.comaws ecr create-repository --repository-name kcl-consumer-app --region us-east-1# tag and push the imagedocker tag kcl-consumer-app:latest $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/kcl-consumer-app:latestdocker push $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/kcl-consumer-app:latest

更新consumer.yaml以包含您剛剛推送到 ECR 的 Docker 映像。清單的其余部分保持不變：

apiVersion: apps/v1kind: Deploymentmetadata: name: kcl-consumerspec: replicas: 1 selector: matchLabels: app: kcl-consumer template: metadata: labels: app: kcl-consumer spec: serviceAccountName: kcl-consumer-app-sa containers: - name: kcl-consumer image: AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/kcl-consumer-app:latest imagePullPolicy: Always env: - name: STREAM_NAME value: kinesis-keda-demo - name: TABLE_NAME value: users - name: APPLICATION_NAME value: kinesis-keda-demo - name: AWS_REGION value: us-east-1 - name: INSTANCE_NAME valueFrom: fieldRef: fieldPath: metadata.name

創(chuàng)建Deployment：

kubectl apply -f consumer.yaml# verify Pod transition to Running statekubectl get pods -w

現(xiàn)在您已經(jīng)部署了消費(fèi)者應(yīng)用程序，KCL庫(kù)應(yīng)該開始行動(dòng)了。它要做的第一件事是在 DynamoDB 中創(chuàng)建一個(gè)“控制表”——這應(yīng)該與 KCL 應(yīng)用程序的名稱相同（在本例中為 ）kinesis-keda-demo。

進(jìn)行初始協(xié)調(diào)和創(chuàng)建表可能需要幾分鐘時(shí)間。您可以檢查消費(fèi)者應(yīng)用程序的日志以跟蹤進(jìn)度。

kubectl logs -f $(kubectl get po -l=app=kcl-consumer --output=jsnotallow={.items..metadata.name})

租約分配完成后，檢查表并記下leaseOwner屬性：

aws dynamodb describe-table --table-name kinesis-keda-demoaws dynamodb scan --table-name kinesis-keda-demo

現(xiàn)在，讓我們使用 AWS CLI 將一些數(shù)據(jù)發(fā)送到 Kinesis 流。

export KINESIS_STREAM=kinesis-keda-demoaws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user1@foo.com --data $(echo -n "{"name":"user1", "city":"new york"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user2@foo.com --data $(echo -n "{"name":"user2", "city":"tel aviv"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user3@foo.com --data $(echo -n "{"name":"user3", "city":"new delhi"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user4@foo.com --data $(echo -n "{"name":"user4", "city":"seattle"}" | base64)

KCL 應(yīng)用程序?qū)⒚織l記錄保存到目標(biāo)DynamoDB表（在本例中已命名users）。您可以檢查表格以驗(yàn)證記錄。

aws dynamodb scan --table-name users

注意到processed_by屬性的值了嗎？它與 KCL 消費(fèi)者相同Pod。這將使我們更容易驗(yàn)證端到端的自動(dòng)縮放過(guò)程。

這是ScaledObject定義。請(qǐng)注意，它的目標(biāo)是kcl-consumer Deployment（我們剛剛創(chuàng)建的那個(gè)）并且shardCount設(shè)置為1：

apiVersion: keda.sh/v1alpha1kind: ScaledObjectmetadata: name: aws-kinesis-stream-scaledobjectspec: scaleTargetRef: name: kcl-consumer triggers: - type: aws-kinesis-stream metadata: # Required streamName: kinesis-keda-demo # Required awsRegion: "us-east-1" shardCount: "1" identityOwner: "operator"

創(chuàng)建KEDAKinesis 縮放器：

kubectl apply -f keda-kinesis-scaler.yaml

Pod我們從我們的一個(gè) KCL 應(yīng)用程序開始。但是，多虧了KEDA，我們現(xiàn)在應(yīng)該看到第二次Pod出現(xiàn)了。

kubectl get pods -l=app=kcl-consumer -w# check logs of the new podkubectl logs -f 

我們的應(yīng)用程序能夠自動(dòng)縮放到兩個(gè)，因?yàn)槲覀冊(cè)诙x中Pods指定了。這意味著Kinesis 流中的每個(gè)分片都會(huì)有一個(gè)。shardCount: "1"ScaledObjectPod

檢查kinesis-keda-demo控制表DynamoDB：您應(yīng)該看到leaseOwner.

讓我們向 Kinesis 流發(fā)送更多數(shù)據(jù)。

export KINESIS_STREAM=kinesis-keda-demoaws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user5@foo.com --data $(echo -n "{"name":"user5", "city":"new york"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user6@foo.com --data $(echo -n "{"name":"user6", "city":"tel aviv"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user7@foo.com --data $(echo -n "{"name":"user7", "city":"new delhi"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user8@foo.com --data $(echo -n "{"name":"user8", "city":"seattle"}" | base64)

驗(yàn)證屬性的值processed_by。由于我們已經(jīng)擴(kuò)展到兩個(gè)Pods，每個(gè)記錄的值應(yīng)該不同，因?yàn)槊總€(gè)記錄都Pod將處理來(lái)自 Kinesis 流的記錄子集。

讓我們將分片數(shù)量從兩個(gè)擴(kuò)展到三個(gè)，并繼續(xù)監(jiān)控KCL應(yīng)用程序的自動(dòng)擴(kuò)展。

aws kinesis update-shard-count --stream-name kinesis-keda-demo --target-shard-count 3 --scaling-type UNIFORM_SCALING

一旦 Kinesis 重新分片完成，KEDA縮放器將開始行動(dòng)并將 KCL 應(yīng)用程序擴(kuò)展到三個(gè)Pods。

kubectl get pods -l=app=kcl-consumer -w

kinesis-keda-demo和之前一樣，在控制表中確認(rèn)Kinesis shard lease已經(jīng)更新DynamoDB。檢查leaseOwner屬性。

繼續(xù)向 Kinesis 流發(fā)送更多數(shù)據(jù)。正如預(yù)期的那樣，Pods 將共享記錄處理，這將反映在表processed_by中的屬性中users。

export KINESIS_STREAM=kinesis-keda-demoaws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user9@foo.com --data $(echo -n "{"name":"user9", "city":"new york"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user10@foo.com --data $(echo -n "{"name":"user10", "city":"tel aviv"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user11@foo.com --data $(echo -n "{"name":"user11", "city":"new delhi"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user12@foo.com --data $(echo -n "{"name":"user12", "city":"seattle"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user14@foo.com --data $(echo -n "{"name":"user14", "city":"tel aviv"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user15@foo.com --data $(echo -n "{"name":"user15", "city":"new delhi"}" | base64)aws kinesis put-record --stream-name $KINESIS_STREAM --partition-key user16@foo.com --data $(echo -n "{"name":"user16", "city":"seattle"}" | base64)

到目前為止，我們只在一個(gè)方向上進(jìn)行了擴(kuò)展。當(dāng)我們減少 Kinesis 流的分片容量時(shí)會(huì)發(fā)生什么？親自嘗試一下：將分片數(shù)從三個(gè)減少到兩個(gè)，看看 KCL 應(yīng)用程序會(huì)發(fā)生什么。

驗(yàn)證端到端解決方案后，您應(yīng)該清理資源以避免產(chǎn)生任何額外費(fèi)用。

刪除 EKS 集群、Kinesis 流和 DynamoDB 表。

eksctl delete cluster --name keda-kinesis-demoaws kinesis delete-stream --stream-name kinesis-keda-demoaws dynamodb delete-table --table-name users

在本文中，您學(xué)習(xí)了如何使用KEDA自動(dòng)縮放KCL使用來(lái)自 Kinesis 流的數(shù)據(jù)的應(yīng)用程序。

您可以根據(jù)您的應(yīng)用要求配置 KEDA 定標(biāo)器。例如，您可以將Kinesis 流中的每三個(gè)分片設(shè)置為shardCount一個(gè)3。Pod然而，如果你想維護(hù)一個(gè)一對(duì)一的映射，你可以設(shè)置為shardCount并1會(huì)KCL處理分布式協(xié)調(diào)和租約分配，從而確保每個(gè)Pod記錄處理器都有一個(gè)實(shí)例。這是一種有效的方法，可讓您擴(kuò)展 Kinesis 流處理管道以滿足應(yīng)用程序的需求。

關(guān)鍵詞：