Kubernetes中的粗粒度并行工作队列处理：推进批处理优化

在现代分布式系统环境中，许多任务涉及批处理，而并行处理将作业划分为可以并行执行的较小部分和单元。Kubernetes以其容器编排能力闻名，提供了一个称为Job的对象来处理此类情况。

并非所有批处理作业都相同。根据任务的细节和性质，它们之间可能需要一些协调；我们可能需要不同的模式来分发这类工作负载。这就是粗粒度并行处理模型的适用场景。

让我们举一个简单的例子：一家面包店每天收到500个定制蛋糕订单，所有面包师一起制作每个蛋糕。突然，他们收到了5000个蛋糕的大订单，如果继续采用原来的方式，每个蛋糕都需要不断的协调和等待共享步骤。我们改为为每个面包师分配整个蛋糕，让他们独立工作而不互相干扰，从而提高任务效率。他们拿一个订单，完成它，然后继续下一个。

这类似于粗粒度并行处理，其中每个工作节点（或Kubernetes中的pod）从队列中获取一个任务，独立处理，然后完成。工作节点之间不需要交互，使其更易于扩展和具备容错能力。

什么是粗粒度并行处理

这种方法允许在多个pod之间分配工作，每个pod处理队列中的单个任务，并实现高效的并行执行。

粗粒度并行的主要优势是：

简单性：任务之间不需要通信
可扩展性：我们可以添加更多工作节点（pod）来提高吞吐量
容错性：一个工作节点的故障不会影响其他节点

在这种模式中，Kubernetes Job中的每个pod执行：

从消息队列（如RabbitMQ）获取单个任务
处理任务
确认完成并退出

这对于可以独立处理的任务很有用。

适用于以下工作负载：

图像或视频处理
数据转换
文件转换
长时间运行的计算

工作队列模型（WQM）概念

为了有效实现粗粒度并行处理，我们主要使用工作队列，这是一个消息系统，任务在其中提交并由工作节点消费。它将任务提交与执行解耦，实现了灵活性和持久性。

这种模式的典型技术栈可能包括：

RabbitMQ、Redis或Amazon SQS作为消息代理
连接队列、拉取任务、处理任务并退出的工作节点应用程序
Kubernetes作业来部署和管理工作节点pod

Kubernetes如何适配

Kubernetes作业允许运行一个或多个pod直至完成。它们非常适合粗粒度并行处理，因为我们可以控制：

要运行的总任务数（completions）
应该并行运行的pod数量（parallelism）
如果pod失败应该发生什么（重启策略）

将作业与工作队列结合意味着每个pod可以：

启动并连接到队列
获取单个任务
执行工作
成功退出

然后Kubernetes标记该pod的完成，并继续启动其他pod，直到所有任务完成。

图1：Kubernetes中的并行处理

设置基础设施

部署消息队列服务

首先部署像RabbitMQ这样的消息队列服务。该服务将保存需要处理的任务。

1
2
3


kubectl create -f https://raw.githubusercontent.com/kubernetes/kubernetes/release-1.3/examples/celery-rabbitmq/rabbitmq-service.yaml

kubectl create -f https://raw.githubusercontent.com/kubernetes/kubernetes/release-1.3/examples/celery-rabbitmq/rabbitmq-controller.yaml

上述命令将创建一个RabbitMQ服务（rabbitmq-service）和一个复制控制器（rabbitmq-controller）。

测试消息队列

现在让我们测试消息队列。我们将创建一个临时交互式pod，安装一些工具，并试验队列。

首先，创建临时交互式pod。

1
2


# 创建临时交互式容器
kubectl run -i --tty temp --image ubuntu:18.04

接下来，安装amqp-tools，以便我们可以使用消息队列。

1
2
3


# 安装一些工具
root@temp:/# apt-get update
root@temp:/# apt-get install -y curl ca-certificates amqp-tools python dnsutils

稍后，我们将制作包含这些软件包的Docker镜像。接下来，我们将检查是否可以发现RabbitMQ服务：

1
2


# 注意rabbitmq-service有一个由Kubernetes提供的DNS名称：
root@temp:/# nslookup rabbitmq-service

如果Kube-DNS设置不正确，上一步可能无法工作。我们还可以在环境变量中找到服务IP：

1

# env | grep RABBIT | grep HOST

RABBITMQ_SERVICE_SERVICE_HOST=10.1.123.32（我们将收到服务的IP地址）。

用任务填充队列

RabbitMQ运行后，用任务填充队列。每个任务可以是队列中的一条消息，例如作业ID或数据处理请求。

我们将验证是否可以创建队列并发布和消费消息。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


# 下一行中，rabbitmq-service是可以访问RabbitMQ服务的主机名。
# 5672是RabbitMQ的标准端口。

root@temp:/# export BROKER_URL=amqp://guest:guest@rabbitmq-service:5672

# 如果您在上一步中无法解析"rabbitmq-service"，
# 则改用此命令：

root@temp:/# export BROKER_URL=amqp://guest:guest@$RABBITMQ_SERVICE_SERVICE_HOST:5672

# 现在创建队列：
root@temp:/# /usr/bin/amqp-declare-queue --url=$BROKER_URL -q test -d test

# 向其发布一条消息：
root@temp:/# /usr/bin/amqp-publish --url=$BROKER_URL -r test -p -b Hello

# 并将其取回：
root@temp:/# /usr/bin/amqp-consume --url=$BROKER_URL -q test -c 1 cat && echo Hello

root@temp:/#

为工作节点创建Docker镜像

构建一个包含以下内容的Docker镜像：

轻量级基础镜像（如Ubuntu）
与RabbitMQ交互所需的工具，如amqp-tools
从队列处理任务的工作节点脚本

定义Kubernetes作业

创建一个Kubernetes作业定义（job.yaml），指定：

完成次数（要处理的总任务数）
并行度（并发运行的pod数量）
要使用的容器镜像
用于连接到消息队列的环境变量

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


apiVersion: batch/v1
kind: Job
metadata:
  name: job-wq-1
spec:
  completions: 8
  parallelism: 2
  template:
    spec:
      containers:
      - name: worker
        image: your-docker-image
        env:
        - name: BROKER_URL
          value: amqp://guest:guest@rabbitmq-service:5672
        - name: QUEUE
          value: job1
      restartPolicy: OnFailure

部署作业

将作业YAML应用到您的Kubernetes：

1

kubectl apply -f job.yaml

监控作业进度：

1

kubectl wait --for=condition=complete --timeout=300s job/job-wq-1

需要遵循的几个考虑因素

Pod生命周期

每个pod处理一个任务并退出。确保提交任务的处理保持不变以处理重试。

队列管理

监控队列以确保任务正在被及时消费和处理。

资源分配

根据队列大小和指定集群中可用的资源调整完成次数和并行度设置。

何时使用此模式

这种粗粒度并行处理模式在以下情况下是理想的：

任务独立且可以隔离处理
有大量小型离散任务
以最小开销实现高吞吐量

结论

在Kubernetes集群中实现粗粒度并行处理模型是一种高效且可扩展的方法，利用了容器自动部署、作业调度和任务负载平衡等编排方法。粗粒度并行处理在处理需要更多计算能力且管理和操作开销最小的任务时非常有效。总之，Kubernetes环境中的粗粒度并行处理模型提供了一个经济高效、优化且具有弹性的架构来并行执行任务。

Kubernetes中的粗粒度并行工作队列处理：优化批处理性能

本文详细介绍了在Kubernetes中实现粗粒度并行处理工作队列的方法，包括消息队列部署、工作节点配置和任务调度策略，帮助优化批处理任务的执行效率和资源利用率。