JSON为角色创建一个包含可信实体的文件：

read -r -d '' TRUST_RELATIONSHIP < trust.json

现在，创建 IAM 角色并附加策略（查看policy.json文件了解详细信息）：

export ROLE_NAME=kcl-consumer-app-roleaws iam create-role --role-name $ROLE_NAME --assume-role-policy-document file://trust.json --description "IRSA for KCL consumer app on EKS"aws iam create-policy --policy-name kcl-consumer-app-policy --policy-document file://policy.jsonaws iam attach-role-policy --role-name $ROLE_NAME --policy-arn=arn:aws:iam::${ACCOUNT_ID}:policy/kcl-consumer-app-policy

关联 IAM 角色和服务帐户：

kubectl annotate serviceaccount -n default kcl-consumer-app-sa eks.amazonaws.com/role-arn=arn:aws:iam::${ACCOUNT_ID}:role/${ROLE_NAME}# verify the annotationkubectl describe serviceaccount/kcl-consumer-app-sa

核心基础设施现已准备就绪。让我们准备并部署消费者应用程序。

将 KCL 消费者应用程序部署到 EKS

您首先需要构建 Docker 镜像并将其推送到Amazon Elastic Container Registry (ECR)（有关Dockerfile详细信息，请参阅 ）。

构建 Docker 镜像并将其推送到 ECR

# 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

创建Deployment：

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

KCL 应用程序自动缩放在 KEDA 中的应用

现在您已经部署了消费者应用程序，KCL库应该开始行动了。它要做的第一件事是在 DynamoDB 中创建一个“控制表”——这应该与 KCL 应用程序的名称相同（在本例中为 ）kinesis-keda-demo。

进行初始协调和创建表可能需要几分钟时间。您可以检查消费者应用程序的日志以跟踪进度。

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

现在，让我们使用 AWS CLI 将一些数据发送到 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 应用程序将每条记录保存到目标DynamoDB表（在本例中已命名users）。您可以检查表格以验证记录。

aws dynamodb scan --table-name users

注意到processed_by属性的值了吗？它与 KCL 消费者相同Pod。这将使我们更容易验证端到端的自动缩放过程。

为 Kinesis 创建 KEDA 定标器

这是ScaledObject定义。请注意，它的目标是kcl-consumer Deployment（我们刚刚创建的那个）并且shardCount设置为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"

创建KEDAKinesis 缩放器：

kubectl apply -f keda-kinesis-scaler.yaml

验证 KCL 应用程序自动缩放

Pod我们从我们的一个 KCL 应用程序开始。但是，多亏了KEDA，我们现在应该看到第二次Pod出现了。

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

我们的应用程序能够自动缩放到两个，因为我们在定义中Pods指定了。这意味着Kinesis 流中的每个分片都会有一个。shardCount: "1"ScaledObjectPod

检查kinesis-keda-demo控制表DynamoDB：您应该看到leaseOwner.

让我们向 Kinesis 流发送更多数据。

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)

验证属性的值processed_by。由于我们已经扩展到两个Pods，每个记录的值应该不同，因为每个记录都Pod将处理来自 Kinesis 流的记录子集。

增加 Kinesis 流容量

让我们将分片数量从两个扩展到三个，并继续监控KCL应用程序的自动扩展。

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

一旦 Kinesis 重新分片完成，KEDA缩放器将开始行动并将 KCL 应用程序扩展到三个Pods。

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

kinesis-keda-demo和之前一样，在控制表中确认Kinesis shard lease已经更新DynamoDB。检查leaseOwner属性。

继续向 Kinesis 流发送更多数据。正如预期的那样，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)

缩小

到目前为止，我们只在一个方向上进行了扩展。当我们减少 Kinesis 流的分片容量时会发生什么？亲自尝试一下：将分片数从三个减少到两个，看看 KCL 应用程序会发生什么。

验证端到端解决方案后，您应该清理资源以避免产生任何额外费用。

删除资源

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

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

结论

在本文中，您学习了如何使用KEDA自动缩放KCL使用来自 Kinesis 流的数据的应用程序。

您可以根据您的应用要求配置 KEDA 定标器。例如，您可以将Kinesis 流中的每三个分片设置为shardCount一个3。Pod然而，如果你想维护一个一对一的映射，你可以设置为shardCount并1会KCL处理分布式协调和租约分配，从而确保每个Pod记录处理器都有一个实例。这是一种有效的方法，可让您扩展 Kinesis 流处理管道以满足应用程序的需求。

本文标题：在Kubernetes上自动缩放KinesisDataStreams应用程序
文章路径：http://www.shufengxianlan.com/qtweb/news40/69340.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联