数据库OLAP大数据后端【免费下载链接】druidApache Druid: a high performance real-time analytics database.项目地址https://gitcode.com/gh_mirrors/druid6/druid点击查看免费下载导读本文面向在 Kubernetes 上部署 Apache Druid 的运维与开发人员系统讲解druid-kubernetes-overlord-extensionsKubernetes Task Scheduling扩展它允许 Overlord 将 Druid 任务直接调度为 Kubernetes Job 运行从而彻底移除 Middle Manager / Indexer 组件实现无 MMMM-less的实时摄入架构。阅读本文你将掌握该扩展的工作原理、Overlord 端完整配置、Pod 适配器Pod Adapter与 Pod 模板选择策略、动态执行配置 API、RBAC 权限要求以及从传统 Middle Manager 集群零停机迁移到 K8s 任务调度的k8sAndWorker混合模式。该扩展的官方完整文档位于 docs/development/extensions-contrib/k8s-jobs.md扩展源码位于 extensions-contrib/kubernetes-overlord-extensions。官方将其标注为 实验性EXPERIMENTAL 功能主要原因是尚未在大量长期运行的 Druid 集群上进行过广泛验证生产使用前请充分评估。工作原理任务如何变成 Kubernetes Job在传统架构中Overlord 将任务派发给 Middle Manager 或 Indexer 上的 worker 进程执行。而本扩展改变了这一模式Overlord 根据指定的 Pod 适配器Pod Adapter为每个任务构建一个 Pod Spec并提交为 Kubernetes Job。每个任务对应一个 K8s JobJob 中的主容器以 internal peon内部 peon方式运行任务。该设计带来的关键特性是任务的天然可恢复性Job 与 Druid 部署本身完全解耦重启 Pod 或进行升级不会影响正在运行的任务任务会继续运行当 Overlord 重新启动后会重新发现并追踪这些 Job。从源码看KubernetesTaskRunnerKubernetesTaskRunner.java是核心执行器其工作方式如下每任务一线程Runner 内部使用容量可配置的线程池Execs.multiThreaded(config.getCapacity(), k8s-task-runner-%d)每个线程负责追踪一个 Job——先调用KubernetesPeonLifecycle把 Job 提交到 K8s再等待其回报成功/失败状态队列排队当线程池没有空闲线程时任务进入队列并保持WAITING状态直到有任务完成腾出线程启动时恢复start()方法会调用client.getPeonJobs()列出正在运行的 Job并通过joinAsync()重新挂接每个 Job 的状态这正是Overlord 重启后继续追踪任务的实现细节源码 KubernetesTaskRunner.java定期清理Runner 内置一个定时清理线程按照taskCleanupInterval周期检查并删除早于taskCleanupDelay的已完成 Job。KubernetesTaskRunnerFactoryKubernetesTaskRunnerFactory.java通过 Guice 在 KubernetesOverlordModule.java 中注册druid.indexer.runner.typek8s绑定到KubernetesTaskRunnerFactorydruid.indexer.runner.typek8sAndWorker绑定到KubernetesAndWorkerTaskRunnerFactory。配置让 Overlord 使用 K8s 任务调度加载扩展首先在 Overlord 进程的扩展加载列表中包含druid-kubernetes-overlord-extensions具体方法参见 加载扩展。核心运行时属性扩展依赖以下关键配置参考 KubernetesTaskRunnerConfig.java 中的字段定义属性说明druid.indexer.runner.type必须设为k8s或迁移模式的k8sAndWorkerdruid.indexer.runner.namespace必填Druid 集群运行所在的 Kubernetes 命名空间druid.indexer.runner.capacity限制同时在飞的 K8s Job 数量默认值为Integer.MAX_VALUEdruid.indexer.task.encapsulatedTask必须设为true几点重要说明capacity 的初始值建议设置为切换 K8s 之前所有 Middle Manager 任务槽位task slots的总和。K8s task runner 为每个创建的 Job 使用一个线程该值设置过大会在 Overlord 上造成内存问题线程数量过多capacity同时作为线程池大小、总槽位数getTotalTaskSlotCount与已用槽位计算的上限源码见 KubernetesTaskRunner.java若完全运行在无 Middle Manager 的集群中还需要设置druid.processing.intermediaryData.storage.typedeepstore让任务中间数据落 deep storage。最小配置示例druid.indexer.runner.typek8s druid.indexer.runner.namespacedefault druid.indexer.runner.capacity10 druid.indexer.task.encapsulatedTasktrue druid.processing.intermediaryData.storage.typedeepstore动态执行配置 API无需重启 OverlordDruid 运维人员可以动态调整扩展的某些特性无需重启 Overlord 服务即可生效。动态配置的核心是 Pod 模板选择Pod Template Selection启用前需要先配置 [Custom Template Pod Adapter](#custom-template-pod-adapter 自定义模板 Pod 适配器)。动态配置对象KubernetesTaskRunnerDynamicConfig在源码中对应配置键k8s.taskrunner.config默认实现为DefaultKubernetesTaskRunnerDynamicConfig默认选择策略为TaskTypePodTemplateSelectStrategy见 KubernetesTaskRunnerDynamicConfig.java。使用这些 API 前请确保拥有资源类型为CONFIG、资源名为CONFIG的读写权限权限说明参见 用户认证与授权。获取动态配置GET /druid/indexer/v1/k8s/taskrunner/executionconfigcurl http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/k8s/taskrunner/executionconfigGET /druid/indexer/v1/k8s/taskrunner/executionconfig HTTP/1.1 Host: http://ROUTER_IP:ROUTER_PORT成功时返回当前动态执行配置的 JSON 对象{ type: default, podTemplateSelectStrategy: { type: selectorBased, selectors: [ { selectionKey: podSpec1, context.tags: { userProvidedTag: [tag1, tag2] }, dataSource: [wikipedia] }, { selectionKey: podSpec2, type: [index_kafka] } ] } }更新动态配置POST /druid/indexer/v1/k8s/taskrunner/executionconfig该端点支持以下可选请求头用于在配置历史中记录变更者与变更说明X-Druid-AuthorString配置变更的作者X-Druid-CommentString本次更新的描述。curl http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/k8s/taskrunner/executionconfig \ --header Content-Type: application/json \ --data { type: default, podTemplateSelectStrategy: { type: selectorBased, selectors: [ { selectionKey: podSpec1, context.tags: { userProvidedTag: [tag1, tag2] }, dataSource: [wikipedia] }, { selectionKey: podSpec2, type: [index_kafka] } ] } }POST /druid/indexer/v1/k8s/taskrunner/executionconfig HTTP/1.1 Host: http://ROUTER_IP:ROUTER_PORT Content-Type: application/json { type: default, podTemplateSelectStrategy: { type: selectorBased, selectors: [ { selectionKey: podSpec1, context.tags: { userProvidedTag: [tag1, tag2] }, dataSource: [wikipedia] }, { selectionKey: podSpec2, type: [index_kafka] } ] } }成功请求返回 HTTP200 OK与空响应体。获取动态配置历史GET /druid/indexer/v1/k8s/taskrunner/executionconfig/history支持以下可选查询参数过滤结果intervalString以 ISO 8601 格式限定的时间区间以/分隔例如2023-07-13/2023-07-19。默认区间为一周可通过在 Coordinator 的runtime.properties中设置druid.audit.manager.auditHistoryMillis调整countInteger只返回最近n条记录。curl http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/k8s/taskrunner/executionconfig/historyGET /druid/indexer/v1/k8s/taskrunner/executionconfig/history HTTP/1.1 Host: http://ROUTER_IP:ROUTER_PORT若没有历史记录则返回空数组示例响应[ { key: k8s.taskrunner.config, type: k8s.taskrunner.config, auditInfo: { author: , comment: , ip: 127.0.0.1 }, payload: {\type\: \default\,\podTemplateSelectStrategy\:{\type\: \taskType\}, auditTime: 2024-06-13T20:59:51.622Z } ]Pod 适配器如何构建 Job 的 Pod SpecPod 适配器决定了 Kubernetes Job 的 Pod 模板如何构建。适配器类型在KubernetesTaskRunnerFactory.buildTaskAdapter()中根据druid.indexer.runner.k8s.adapter.type属性选择源码 KubernetesTaskRunnerFactory.java。Overlord Single Container Pod Adapter该适配器直接取 Overlord Pod 的 podSpec据此创建 Kubernetes Job是默认的 Pod 适配器实现。显式启用方式druid.indexer.runner.k8s.adapter.type: overlordSingleContainerOverlord Multi Container Pod Adapter该适配器同样基于 Overlord Pod 的 podSpec 创建 Job但会使用 kubexit 管理主容器运行 Druid peon 的容器与 Overlord Pod Spec 中定义的其它 sidecar 之间的依赖顺序。如果你的 sidecar 是 Splunk、Istio 等该适配器可以正确处理它们。启用方式druid.indexer.runner.k8s.adapter.type: overlordMultiContainerMulti Container 适配器的关键限制——必须显式声明 command要让 sidecar 支持正常工作Docker 镜像中的入口点/命令必须在 Pod Spec 中显式定义否则扩展无法解析。以下写法不可行ENTRYPOINT: [foo.sh]container: name: foo args: - arg1 - arg2因为扩展无法推断 command 是什么。即使是 Istio 这类由 service mesh 动态注入的 sidecar 也同样需要显式声明。正确写法是在 sidecar spec 中显式给出 commandcontainer: name: foo command: foo.sh args: - arg1 - arg2Dockerfile 可以保持不变。对于上述两个适配器可以通过以下属性为 K8s Job/Pod 添加可选标签与注解值均为 JSON 对象字符串druid.indexer.runner.labels: {key:value} druid.indexer.runner.annotations: {key:value}迁移注意原先为 Middle Manager 任务配置的所有其它配置都需要迁移到 Overlord 下且 javaOpts 必须写成数组形式druid.indexer.runner.javaOptsArraydruid.indexer.runner.javaOpts已不再支持。Custom Template Pod Adapter自定义模板 Pod 适配器自定义模板适配器允许按任务类型指定 Pod 模板文件以更灵活地定义 Pod。它要求 Overlord 文件系统上存在 Pod Template 文件该模板作为 K8s Job Pod Spec 的基础可覆盖 labels、环境变量、资源、注解甚至基础镜像。启用方式druid.indexer.runner.k8s.adapter.type: customTemplateAdapter druid.indexer.runner.k8s.podTemplate.base: /path/to/basePodSpec.yaml示例 Pod Template以下示例使用常规的 druid docker 镜像apiVersion: v1 kind: PodTemplate template: metadata: annotations: sidecar.istio.io/proxyCPU: 512m # to handle a injected istio sidecar labels: app.kubernetes.io/name: druid-realtime-backend spec: affinity: {} containers: - command: - sh - -c - | /peon.sh /druid/data 1 env: - name: CUSTOM_ENV_VARIABLE value: hello image: apache/druid:{{DRUIDVERSION}} name: main ports: - containerPort: 8091 name: druid-tls-port protocol: TCP - containerPort: 8100 name: druid-port protocol: TCP resources: limits: cpu: 1 memory: 2400M requests: cpu: 1 memory: 2400M volumeMounts: - mountPath: /opt/druid/conf/druid/cluster/master/coordinator-overlord # runtime props are still mounted in this location because thats where peon.sh looks for configs name: nodetype-config-volume readOnly: true - mountPath: /druid/data name:>kind: ConfigMap metadata: name: druid-tiny-cluster-peons-config namespace: default apiVersion: v1 data: jvm.config: |- -server -XX:MaxDirectMemorySize1000M -Duser.timezoneUTC -Dfile.encodingUTF-8 -Dlog4j.debug -Djava.util.logging.managerorg.apache.logging.log4j.jul.LogManager -Djava.io.tmpdir/druid/data -Xmx1024M -Xms1024M log4j2.xml: |- ?xml version1.0 encodingUTF-8 ? Configuration statusWARN Appenders Console nameConsole targetSYSTEM_OUT PatternLayout pattern%d{ISO8601} %p [%t] %c - %m%n/ /Console /Appenders Loggers Root levelinfo AppenderRef refConsole/ /Root /Loggers /Configuration runtime.properties: | druid.port8100 druid.servicedruid/peon druid.server.http.numThreads5 druid.indexer.task.baseTaskDir/druid/data druid.indexer.runner.typek8s druid.peon.moderemote druid.indexer.task.encapsulatedTasktruePod 模板选择策略Custom Template Pod Adapter 可以使用动态执行配置来决定某个任务该用哪个 Pod 模板。基于任务类型选择TaskTypePodTemplateSelectStrategy该策略根据任务类型选择 Pod 模板是默认的 Pod 模板选择策略。对应源码 TaskTypePodTemplateSelectStrategy.java若模板集合中存在以任务类型为 key 的模板则使用它否则回退到base模板。显式指定该策略即动态配置中的默认值{ type: default }任务类型专属的 Pod 模板通过运行时属性指定{taskType}为任务类型名称例如index_paralleldruid.indexer.runner.k8s.podTemplate.{taskType}: /path/to/taskSpecificPodSpec.yaml环境变量转义说明如果使用默认镜像的环境变量解析特性来设置运行时属性指定 Pod 模板时需要多加一层转义下划线。例如设置运行时属性druid.indexer.runner.k8s.podTemplate.index_kafka时环境变量应写为druid_indexer_runner_k8s_podTemplate_index__kafkaindex_kafka中的下划线转义为双下划线。基于任务类型的模板选择配置示例druid.indexer.runner.k8s.podTemplate.base/path/to/basePodSpec.yaml druid.indexer.runner.k8s.podTemplate.index_kafka/path/to/kafkaPodSpec.yaml基于一个或多个条件选择SelectorBasedPodTemplateSelectStrategy该策略评估selectors中的一系列条件来决定使用哪个 Pod 模板运行任务。Pod 模板在运行时属性中按druid.indexer.runner.k8s.podTemplate.selectionKey...配置。对应源码为 SelectorBasedPodTemplateSelectStrategy.java 与 Selector.java。{ type: selectorBased, selectors: [ { selectionKey: podSpec1, context.tags: { userProvidedTag: [tag1, tag2] }, dataSource: [wikipedia] }, { selectionKey: podSpec2, type: [index_kafka] } ] }选择规则按顺序处理Selectors 按顺序求值Druid 选择第一个匹配的 selector 对应的模板兜底回退任务不匹配任何 selector 时使用basePod 模板条件全匹配selector 内的所有条件都必须满足才算匹配。一个 selector 可以匹配type任务的类型dataSource任务的目标数据源context.tags任务 context 中传入的标签。从 Selector.java 的evaluate()实现可以看到context.tags条件要求任务 context 中tags字段非空且指定标签 key 的值命中条件值集合type与dataSource条件均为集合包含判断任一条件不满足即不匹配。完整示例先设置运行时属性定义可供选择的 Pod Specdruid.indexer.runner.k8s.podTemplate.base/path/to/basePodSpec.yaml druid.indexer.runner.k8s.podTemplate.podSpec1/path/to/podSpecWithHighMemRequests.yaml druid.indexer.runner.k8s.podTemplate.podSpec2/path/to/podSpecWithLowCpuRequests.yaml再通过动态执行配置定义选择策略{ type: default, podTemplateSelectStrategy: { type: selectorBased, selectors: [ { selectionKey: podSpec1, context.tags: { userProvidedTag: [tag1, tag2] }, dataSource: [wikipedia] }, { selectionKey: podSpec2, type: [index_kafka] } ] } }Druid 将按如下规则选择模板当以下两个条件同时满足时使用podSpecWithHighMemRequests.yaml任务 context 包含 key 为userProvidedTag、值为tag1或tag2的标签任务面向wikipedia数据源。任务类型为index_kafka时使用podSpecWithLowCpuRequests.yaml其余任务一律使用basePodSpec.yaml。注意若存在一个针对wikipedia数据源、带userProvidedTag: tag1标签的index_kafka任务由于第一个 selector 先匹配Druid 会选择podSpecWithHighMemRequests.yaml而非 podSpec2。完整属性参考下表为 K8s task runner 的全部运行时属性默认值与约束可对照源码 KubernetesTaskRunnerConfig.java 验证属性可选值说明默认值必填druid.indexer.runner.debugJobsboolean任务完成后是否清理 K8s jobs。false否druid.indexer.runner.sidecarSupportboolean已废弃请改用运行时属性druid.indexer.runner.k8s.adapter.type: overlordMultiContainer指定适配器类型。若 Overlord Pod 有 sidecar该选项将尝试用与 Overlord Pod 相同的 sidecar 启动任务。false否druid.indexer.runner.primaryContainerNameString使用 sidecar 运行时primaryContainerName应设为主 Druid 容器名如druid-overlord。若不设置则假定 podSpec 列表中的第一个容器为主容器——但当 Istio 等 service mesh 动态注入 sidecar 时istio-proxy 可能被放在第一位因此建议显式指定。podSpec 列表中的第一个容器否druid.indexer.runner.kubexitImageString使用 kubexit 项目在主 Pod 完成后协助关闭 sidecar否则带 sidecar 的 Job 永不终止。karlkfi/kubexit:v0.3.2否druid.indexer.runner.disableClientProxyboolean若存在全局 http(s) 代理且希望绕过时使用。false否druid.indexer.runner.maxTaskDurationDuration任务允许运行的最长时间超过即被杀死。PT4H否druid.indexer.runner.taskCleanupDelayDurationJob 在 K8s 中保留多久后被回收。P2D否druid.indexer.runner.taskCleanupIntervalDuration多久检查一次待回收的 Job。PT10M否druid.indexer.runner.K8sjobLaunchTimeoutDuration启动 K8s 任务前等待多久超时标记为失败资源受限的集群上可能需要更长等待。PT1H否druid.indexer.runner.javaOptsArrayJsonArray任务的 java opts。-Xmx1g否druid.indexer.runner.labelsJsonObject要添加到 peon Pod 的额外标签。{}否druid.indexer.runner.annotationsJsonObject要添加到 peon Pod 的额外注解。{}否druid.indexer.runner.peonMonitorsJsonArray覆盖druid.monitoring.monitors。如果不希望从 Overlord 继承 monitors 时使用该属性。[]否druid.indexer.runner.graceTerminationPeriodSecondsLong收到 sigterm 后等待容器生命周期钩子完成的秒数。若希望任务更短时间持有锁请设置较小值。PT30SK8s 默认值否druid.indexer.runner.capacityInteger可同时发送到 Kubernetes 的并发 Job 数量。2147483647否druid.indexer.runner.cpuCoreInMicroInteger任务的 CPU 微核数。1000否补充说明源码层面disableClientProxy在 KubernetesOverlordModule.java 的makeKubernetesClient()中生效——当其为 true 时构建 fabric8 KubernetesConfig并将 http/https 代理置空。peonMonitors的设计动机是ForkingTaskRunner 从 MM 继承 monitors而在 k8s 模式下 peon 会从 Overlord 继承 monitors如果 Overlord 配置了例如TaskCountStatsMonitorpeon 进程可能因无法注入该 monitor 而失败。新增指标指标说明维度正常值k8s/peon/startup/timepeon Pod 启动所需毫秒数。dataSource,taskId,taskType,groupId,taskStatus,tags视环境而定注意事项Gotchas统一命名空间属于同一个 Druid 集群的所有 Druid Pod 必须位于同一个 Kubernetes 命名空间内RBAC 权限Overlord 的 service account 必须具备与 Kubernetes 交互所需的 role binding。示例如下kind: Role apiVersion: rbac.authorization.k8s.io/v1 metadata: namespace: druid-namespace name: druid-k8s-task-scheduler rules: - apiGroups: [batch] resources: [jobs] verbs: [get, watch, list, delete, create] - apiGroups: [] resources: [pods, pods/log] verbs: [get, watch, list, delete, create] --- kind: RoleBinding apiVersion: rbac.authorization.k8s.io/v1 metadata: name: druid-k8s-binding namespace: druid-namespace subjects: - kind: ServiceAccount name: druid-overlord-k8s-service-account namespace: druid-namespace roleRef: kind: Role name: druid-k8s-task-scheduler apiGroup: rbac.authorization.k8s.io从权限列表可以看出 Overlord 需要管理batch/jobs以及pods、pods/log资源这正对应KubernetesPeonClient提交 Job、查看日志与清理 Job 的底层操作。迁移模式Kubernetes and Worker Task Runner零停机迁移如果集群当前仍有任务运行在 Middle Manager 或 Indexer 上并希望零停机迁移到无 MM 摄入可以使用迁移模式该模式能够同时从 Middle Manager/Indexer 与 Kubernetes 读取任务并将任务写入 Middle Manager 或 Kubernetes 中的任意一方。启用方式将druid.indexer.runner.type: k8s替换为druid.indexer.runner.type: k8sAndWorker迁移模式附加配置属性可选值说明默认值必填druid.indexer.runner.k8sAndWorker.runnerStrategy.typeString如k8s、worker、taskType定义任务运行器的选择策略。k8s否druid.indexer.runner.k8sAndWorker.runnerStrategy.workerTypeString如httpRemote、remote指定 worker task runner 的变体。httpRemote否taskType策略下的配置druid.indexer.runner.k8sAndWorker.runnerStrategy.taskType.defaultString如k8s、worker无覆盖规则适用时使用的默认 runner确保始终存在兜底 runner。无否druid.indexer.runner.k8sAndWorker.runnerStrategy.taskType.overridesJsonObject如{index_kafka: worker}按任务类型定义 runner 的覆盖规则实现细粒度控制。{}否源码层面的实现要点见 KubernetesAndWorkerTaskRunnerConfig.java 与 KubernetesOverlordModule.javaworkerType只允许httpRemoteHttpRemoteTaskRunnerFactory基于 HTTP 的远程任务运行器不依赖 Zookeeper或remoteRemoteTaskRunnerFactory基于 Zookeeper二者之一配置其它值会在启动时抛出校验异常runnerStrategy.type由RunnerStrategyProvider动态注入它读取druid.indexer.runner.k8sAndWorker.runnerStrategy.{strategy}前缀下的配置并通过JsonConfigProvider构建对应的RunnerStrategy实现KubernetesRunnerStrategy/WorkerRunnerStrategy/TaskTypeRunnerStrategy位于 runnerstrategy 包迁移模式下 Overlord 需要同时具备两套 runner 的组件KubernetesTaskRunnerFactory与HttpRemoteTaskRunnerFactory/RemoteTaskRunnerFactory均已在模块中注册。总结druid-kubernetes-overlord-extensions为 Kubernetes 上的 Druid 提供了无 Middle Manager的摄入执行路径任务以 K8s Job 形式运行天然与 Druid 部署解耦并可在 Overlord 重启后恢复通过三种 Pod 适配器单容器、多容器、自定义模板灵活定义 Job 的 Pod Spec通过动态执行配置 API 在线调整 Pod 模板选择策略k8sAndWorker混合模式则支撑从传统 Middle Manager 集群零停机迁移。在使用时请特别注意capacity与线程/内存的关系、多容器适配器必须显式声明 command、统一命名空间与 RBAC 权限等约束。延伸阅读扩展的完整官方文档扩展 READMEKubernetesOverlordModule 源码模块与 Guice 绑定KubernetesTaskRunner 源码任务执行与恢复核心KubernetesTaskRunnerConfig 源码全部运行时属性动态配置与 Pod 模板选择策略实现测试用例目录覆盖配置解析、适配器与选择策略扩展加载方式Kubernetes 上的集群部署指南赞分享数据库OLAP大数据后端【免费下载链接】druidApache Druid: a high performance real-time analytics database.项目地址https://gitcode.com/gh_mirrors/druid6/druid点击查看免费下载相关推荐Apache Druid任务管理详解Overlord与任务优先级调度机制Apache Druid任务管理详解Overlord与任务优先级调度机制 在实时数据分析场景中任务调度的效率直接影响数据处理的及时性和系统资源利用率。Apa数据库数据分析OLAP大数据实时分析数据仓库后端Kubernetes Job 批处理任务详解从 Job Spec 到 Bare Pod 替代实战Kubernetes Job 批处理任务详解从 Job Spec 到 Bare Pod 替代实战 Job 是 Kubernetes 中负责批处理任务Batc教程云原生容器编排Apache DolphinScheduler K8S 节点任务基于 Kubernetes Job 的批量任务编排实践Apache DolphinScheduler K8S 节点任务基于 Kubernetes Job 的批量任务编排实践 Apache DolphinSched任务调度大数据后端前端创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考