在 Kubernetes 上自动扩展 DynamoDB 流应用

你将看到< a rel =“noopener noreferrer” target=“_ blank”> DynamoDB Streams scaler正在动作，它是根据DynamoDB Stream的分片数量而扩展的。

现在让我们继续这个教程的实际部分。

必备条件

除了AWS账户外，您还需要安装 < a rel =“noopener noreferrer” target=“_ blank”> AWS CLI ，< a rel =“noopener noreferrer” target=“_ blank”> kubectl 和< a> Docker 。

设置EKS集群并创建DynamoDB表

有多种方法可以创建< a rel =“noopener noreferrer” target=“_ blank”> Amazon EKS集群 。我更喜欢使用< a> eksctl CLI，因为它提供的便利。使用< code> eksctl 创建EKS集群可能就像这样简单：

eksctl create cluster --name <cluster name> --region <region e.g. us-east-1>

< br />

有关详细信息，请参阅< a rel =“noopener noreferrer” target=“_ blank”>开始使用Amazon EKS - eksctl。

创建启用了流的DynamoDB表来持久保存应用程序数据并访问更改数据源。您可以使用AWS CLI使用以下命令创建表：

aws dynamodb create-table \
    --table-name users \
    --attribute-definitions AttributeName=email,AttributeType=S \
    --key-schema AttributeName=email,KeyType=HASH \
    --billing-mode PAY_PER_REQUEST \
    --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

< br />

我们需要创建另一个表，它将用作第一个表的副本。

aws dynamodb create-table \
    --table-name users_replica \
    --attribute-definitions AttributeName=email,AttributeType=S \
    --key-schema AttributeName=email,KeyType=HASH \
    --billing-mode PAY_PER_REQUEST

< br />

克隆此GitHub存储库并将其更改为正确的目录：

git clone https://github.com/abhirockzz/dynamodb-streams-keda-autoscale
cd dynamodb-streams-keda-autoscale

< br />

好的，让我们开始吧！

在EKS上设置和配置KEDA

本教程中，您将使用< a rel =“noopener noreferrer” target=“_ blank”>YAML文件来部署 KEDA ，但您也可以使用< a rel =“noopener noreferrer” target=“_ blank”>Helm图表。

安装 KEDA ：

# update version 2.8.2 if required

kubectl apply -f https://github.com/kedacore/keda/releases/download/v2.8.2/keda-2.8.2.yaml

< br />

验证安装：

# check Custom Resource Definitions
kubectl get crd

# check KEDA Deployments
kubectl get deployment -n keda

# check KEDA operator logs
kubectl logs -f $(kubectl get pod -l=app=keda-operator -o jsonpath='{.items[0].metadata.name}' -n keda) -n keda

< br />

配置IAM角色

KEDA运算符以及DynamoDB流消费者应用程序需要调用AWS API。由于两者将作为EKS中的< code> Deployment 运行，因此我们将使用< a rel =“noopener noreferrer” target=“_ blank”>服务帐户的IAM角色（IRSA）提供所需的权限。

在我们特定的情况下：

• < code> KEDA 运算符需要能够获取有关DynamoDB表和流的信息
• 应用程序（特指KCL 1.x库）需要与Kinesis和DynamoDB交互 - 它需要一些< a rel =“noopener noreferrer” target=“_ blank”> IAM权限这样做。

为KEDA Operator配置IRSA

将您的AWS帐户ID和OIDC身份提供者设置为环境变量：

ACCOUNT_ID=$(aws sts get-caller-identity --query "Account" --output text)

#update the cluster name and region as required
export EKS_CLUSTER_NAME=demo-eks-cluster
export AWS_REGION=us-east-1

OIDC_PROVIDER=$(aws eks describe-cluster --name $EKS_CLUSTER_NAME --query "cluster.identity.oidc.issuer" --output text | sed -e "s/^https:\/\///")

< br />

创建包含角色的受信任实体的< code> JSON 文件：

read -r -d '' TRUST_RELATIONSHIP <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::${ACCOUNT_ID}:oidc-provider/${OIDC_PROVIDER}"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "StringEquals": {
          "${OIDC_PROVIDER}:aud": "sts.amazonaws.com",
          "${OIDC_PROVIDER}:sub": "system:serviceaccount:keda:keda-operator"
        }
      }
    }
  ]
}
EOF
echo "${TRUST_RELATIONSHIP}" > trust_keda.json

< br />

现在，创建IAM角色并附加策略（查看< code> policy_dynamodb_streams_keda.json 文件以获取详细信息）：

export ROLE_NAME=keda-operator-dynamodb-streams-role

aws iam create-role --role-name $ROLE_NAME --assume-role-policy-document file://trust_keda.json --description "IRSA for DynamoDB streams KEDA scaler on EKS"

aws iam create-policy --policy-name keda-dynamodb-streams-policy --policy-document file://policy_dynamodb_streams_keda.json
aws iam attach-role-policy --role-name $ROLE_NAME --policy-arn=arn:aws:iam::${ACCOUNT_ID}:policy/keda-dynamodb-streams-policy

< br />

关联IAM角色和服务帐户：

kubectl annotate serviceaccount -n keda keda-operator eks.amazonaws.com/role-arn=arn:aws:iam::${ACCOUNT_ID}:role/${ROLE_NAME}

# verify the annotation 
kubectl describe serviceaccount/keda-operator -n keda

< br />

您需要重新启动< code> KEDA 运算符< code> Deployment 才能使其生效：

kubectl rollout restart deployment.apps/keda-operator -n keda

# to verify, confirm that the KEDA operator has the right environment variables
kubectl describe pod -n keda $(kubectl get po -l=app=keda-operator -n keda --output=jsonpath={.items..metadata.name}) | grep "^\s*AWS_"

# expected output

AWS_STS_REGIONAL_ENDPOINTS:   regional
AWS_DEFAULT_REGION:           us-east-1
AWS_REGION:                   us-east-1
AWS_ROLE_ARN:                 arn:aws:iam::<AWS_ACCOUNT_ID>:role/keda-operator-dynamodb-streams-role
AWS_WEB_IDENTITY_TOKEN_FILE:  /var/run/secrets/eks.amazonaws.com/serviceaccount/token

< br />

为DynamoDB流消费应用程序配置IRSA

首先创建一个Kubernetes服务帐户：

kubectl apply -f - <<EOF
apiVersion: v1
kind: ServiceAccount
metadata:
  name: dynamodb-streams-consumer-app-sa
EOF

< br />

创建包含角色的受信任实体的< code> JSON 文件：

read -r -d '' TRUST_RELATIONSHIP <<EOF
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::${ACCOUNT_ID}:oidc-provider/${OIDC_PROVIDER}"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "StringEquals": {
          "${OIDC_PROVIDER}:aud": "sts.amazonaws.com",
          "${OIDC_PROVIDER}:sub": "system:serviceaccount:default:dynamodb-streams-consumer-app-sa"
        }
      }
    }
  ]
}
EOF
echo "${TRUST_RELATIONSHIP}" > trust.json

< br />

现在，创建IAM角色并附加策略。更新< code> policy.json 文件，并输入区域和AWS帐户详细信息。

export ROLE_NAME=dynamodb-streams-consumer-app-role

aws iam create-role --role-name $ROLE_NAME --assume-role-policy-document file://trust.json --description "IRSA for DynamoDB Streams consumer app on EKS"

aws iam create-policy --policy-name dynamodb-streams-consumer-app-policy --policy-document file://policy.json

aws iam attach-role-policy --role-name $ROLE_NAME --policy-arn=arn:aws:iam::${ACCOUNT_ID}:policy/dynamodb-streams-consumer-app-policy

< br />

关联IAM角色和服务帐户：

kubectl annotate serviceaccount -n default dynamodb-streams-consumer-app-sa eks.amazonaws.com/role-arn=arn:aws:iam::${ACCOUNT_ID}:role/${ROLE_NAME}

# verify the annotation
kubectl describe serviceaccount/dynamodb-streams-consumer-app-sa

< br />

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

将DynamoDB流消费应用程序部署到EKS

我们首先需要构建Docker镜像并将其推送到ECR（您可以参考< code> Dockerfile 以获取详细信息）。

构建并推送Docker镜像至ECR

# create runnable JAR file
mvn clean compile assembly\:single

# build docker image
docker build -t dynamodb-streams-consumer-app .

AWS_ACCOUNT_ID=$(aws sts get-caller-identity --query "Account" --output text)

# create a private ECR repo
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com

aws ecr create-repository --repository-name dynamodb-streams-consumer-app --region us-east-1

# tag and push the image
docker tag dynamodb-streams-consumer-app:latest $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/dynamodb-streams-consumer-app:latest
docker push $AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/dynamodb-streams-consumer-app:latest

< br />

部署消费者应用程序

更新< code> consumer.yaml 以包括您刚刚推送到ECR的Docker镜像和源表的DynamoDB流的ARN。清单的其余部分保持不变。

要检索流的ARN，请运行以下命令：

aws dynamodb describe-table --table-name users | jq -r '.Table.LatestStreamArn'

< br />

consumer.yaml Deployment 清单如下：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: dynamodb-streams-kcl-consumer-app
spec:
  replicas: 1
  selector:
    matchLabels:
      app: dynamodb-streams-kcl-consumer-app
  template:
    metadata:
      labels:
        app: dynamodb-streams-kcl-consumer-app
    spec:
      serviceAccountName: dynamodb-streams-kcl-consumer-app-sa
      containers:
        - name: dynamodb-streams-kcl-consumer-app
          image: AWS_ACCOUNT_ID.dkr.ecr.us-east-1.amazonaws.com/dynamodb-streams-kcl-consumer-app:latest
          imagePullPolicy: Always
          env:
            - name: TARGET_TABLE_NAME
              value: users_replica
            - name: APPLICATION_NAME
              value: dynamodb-streams-kcl-app-demo
            - name: SOURCE_TABLE_STREAM_ARN
              value: <enter ARN>
            - name: AWS_REGION
              value: us-east-1
            - name: INSTANCE_NAME
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name

< br />

创建 Deployment ：

kubectl apply -f consumer.yaml

# verify Pod transition to Running state
kubectl get pods -w

< br />

DynamoDB Streams消费者应用程序自动扩展与KEDA

现在您已经部署了消费者应用程序，KCL适配器库应该开始工作。它将首先在DynamoDB中创建一个“控制表” - 它应与应用程序的名称相同（在本例中为< code> dynamodb-streams-kcl-app-demo ）。

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

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

完成租约分配后，请检查表并注意leaseOwner属性：

aws dynamodb describe-table --table-name dynamodb-streams-kcl-app-demo

向DynamoDB表添加数据

现在，您已部署消费者应用程序，让我们向源DynamoDB表（users）添加数据。

您可以使用producer.sh脚本。

export export TABLE_NAME=users
./producer.sh

检查消费者日志以查看消息的处理情况：

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

检查目标表（users_replica）以确认DynamoDB流消费者应用程序实际上已经复制了数据。

aws dynamodb scan --table-name users_replica

注意processed_by属性的值？它与消费者应用程序Pod相同。这将使我们更容易验证端到端的自动扩展过程。

创建KEDA定标器

使用定标器定义：

kubectl apply -f keda-dynamodb-streams-scaler.yaml

这是ScaledObject定义。请注意，它针对dynamodb-streams-kcl-consumer-appDeployment（我们刚刚创建的）进行定位，shardCount设置为2：

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name:  aws-dynamodb-streams-scaledobject
spec:
  scaleTargetRef:
    name: dynamodb-streams-kcl-consumer-app
  triggers:
  - type: aws-dynamodb-streams
    metadata:
      awsRegion: us-east-1
      tableName: users
      shardCount: "2"
      identityOwner: "operator"

关于shardCount属性的说明：

我们使用了2的shardCount值。非常重要的一点是我们使用了支持"每个shard最多同时支持2个消费者"的DynamoDB Streams Kinesis适配器库，它使用支持KCL 1.x。这意味着您不能有超过两个消费者应用程序实例处理相同的DynamoDB流shard。

但是，这个KEDA定标器配置将确保每两个shards就有一个Pod。因此，例如，如果有四个shards，该应用程序将扩展到两个Pods。如果有六个shards，将有三个Pods，依此类推。当然，您可以选择每个shard一个Pod，只需将shardCount设置为1。

要跟踪DynamoDB流中的shard数量，可以运行以下命令：

aws dynamodbstreams describe-stream --stream-arn $(aws dynamodb describe-table --table-name users | jq -r '.Table.LatestStreamArn') | jq -r '.StreamDescription.Shards | length'

我还使用了一个叫做jq的实用程序。如果您想要shards的详细信息：

aws dynamodbstreams describe-stream --stream-arn $(aws dynamodb describe-table --table-name users | jq -r '.Table.LatestStreamArn') | jq -r '.StreamDescription.Shards'

验证DynamoDB流消费者应用程序自动扩展

我们从一个应用程序的Pod开始。但是，由于KEDA，我们应该看到额外的Pod自动出现以满足消费者应用程序的处理要求。

为了确认，请检查Pods的数量：

kubectl get pods -l=app=dynamodb-streams-kcl-consumer-app-consumer

很可能您会看到DynamoDB流中有四个shards和两个Pod。根据将数据产生到DynamoDB表的速率，这可能会发生变化（增加/减少）。

与以前一样，验证DynamoDB目标表（users_replica）中的数据，并注意processed_by属性。因为我们已经扩展到额外的Pods，每个Pod都将处理DynamoDB更改流的一部分消息，所以每条记录的值都应该不同。

另外，确保检查DynamoDB中dynamodb-streams-kcl-app-demo控制表。您应该看到leaseOwner的更新，反映现在有两个Pods从DynamoDB流中消费。

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

删除资源

删除EKS集群和DynamoDB表。

eksctl delete cluster --name <enter cluster name>
aws dynamodb delete-table --table-name users
aws dynamodb delete-table --table-name users_replica

结论

您应该尝试的用例：

• 进一步扩展 - 如何让DynamoDB流增加其分片数？当分片数增加时，消费者实例Pod的数量会发生什么变化？
• 缩小规模 - 当DynamoDB流的分片容量减少时会发生什么情况？

在本文中，我们演示了如何使用KEDA和DynamoDB流，结合了两种强大的技术（变更数据捕获和自动扩展），来构建可伸缩的、事件驱动的系统，以根据应用程序的数据处理需求进行动态调整。