实战分享：DolphinScheduler 中 Shell 使命环境变量最佳设置方式 ...

在使用 Apache DolphinScheduler 编排使命的过程中，Shell 类型使命是最常见的使命类型之一。然而，很多用户在实际使用中都会遇到一个看似简朴却常常引发问题的问题——环境变量怎么设置才有效？
如果你也曾经因为使命执行环境差别等、找不到命令路径、引用变量失败等问题而抓狂，那么这篇文章将为你拨开迷雾。本文将深入解析 DolphinScheduler 中 Shell 使命的环境变量设置机制，分享几种常见的设置方式、注意事项以及实战踩坑履历，帮助你高效、稳定地设置使命运行环境。
使命类型总结

• SHELL使命类型:
  SHELL、JAVA、PYTHON、FLINK、MR、FLINK_STREAM、HIVECLI、SPARK、SEATUNNEL、DATAX、SQOOP、DATA_QUALICY、JUPYTER、MLFLOW、OPENMLDB、DVC、PYTORCH、KUBEFLOW、CHUNJUN、LINKIS
  

注意 : 所谓的SHELL使命类型，都是对SHELL使命类型举行的封装，说白了底层调用的就是Java ProcessBudiler封装的SHELL。

• SQL使命类型(JDBC):
  SQL、PROCEDURE
  

注意 : SQL使命类型实在使用的就是各个DB驱动的JDBC举行的操作。

• HTTP使命类型:
  HTTP、DINKY、PIGEON(WebSocket)
  

注意 : HTTP使命类型实在就是访问其OPEN API，举行HttpClient封装调用的操作。

• 逻辑节点:
  SUB_PROCESS、DEPENDENT、CONDITIONS、SWITHC、DYNAMIC
  

注意 : 所谓的逻辑节点是虚拟使命，这类使命不会调治到Worker节点上去运行，只会在Master节点作为控制节点。

• Client使命类型:
  EMR、K8S、DMS、DATA_FACTORY、SAGEMAKER、ZEPPELIN、DATASYNC、REMOTESHELL
  

注意 : 实在就是调用各个使命的开放的Client举利用命的封装。
Shell使命怎么设置环境变量呢？

因为可能涉及到一个组件的差别的版本的客户端，比如说Spark2、Spark3。还有就是针对差别集群的差别客户端，比如说集群1的Spark3客户端和集群2的Spark客户端。 像这样的需求，怎么在dolphinscheduler中举行设置呢？大概说有几种设置方式呢？
两种方式 : 1、通过task差别的环境变量 2、默认的环境变量
1. 通过task差别的环境变量

安全中心 -> 环境管理

使命中引用

默认的环境变量

common.properties

1. # The default env list will be load by Shell task, e.g. /etc/profile,~/.bash_profile
2. shell.env_source_list=/etc/profile
3. # The interceptor type of Shell task, e.g. bash, sh, cmd
4. shell.interceptor.type=bash

复制代码

org.apache.dolphinscheduler.plugin.task.api.shell.ShellInterceptorBuilderFactory

1. public class ShellInterceptorBuilderFactory {
3.     private final static String INTERCEPTOR_TYPE = PropertyUtils.getString("shell.interceptor.type", "bash");
5.     @SuppressWarnings("unchecked")
6.     public static IShellInterceptorBuilder newBuilder() {
7.         // TODO 默认的走的是这个逻辑
8.         if (INTERCEPTOR_TYPE.equalsIgnoreCase("bash")) {
9.             return new BashShellInterceptorBuilder();
10.         }
11.         if (INTERCEPTOR_TYPE.equalsIgnoreCase("sh")) {
12.             return new ShShellInterceptorBuilder();
13.         }
14.         if (INTERCEPTOR_TYPE.equalsIgnoreCase("cmd")) {
15.             return new CmdShellInterceptorBuilder();
16.         }
17.         throw new IllegalArgumentException("not support shell type: " + INTERCEPTOR_TYPE);
18.     }
20. }

复制代码

org.apache.dolphinscheduler.plugin.task.api.shell.bash.BashShellInterceptorBuilder

1. public class BashShellInterceptorBuilder
2.         extends
3.             BaseLinuxShellInterceptorBuilder<BashShellInterceptorBuilder, BashShellInterceptor> {
5.     @Override
6.     public BashShellInterceptorBuilder newBuilder() {
7.         return new BashShellInterceptorBuilder();
8.     }
10.     @Override
11.     public BashShellInterceptor build() throws FileOperateException, IOException {
12.         // TODO 这里是生成shell脚本的核心点
13.         generateShellScript();
14.         List<String> bootstrapCommand = generateBootstrapCommand();
15.         // TODO 实例化BashShellInterceptor
16.         return new BashShellInterceptor(bootstrapCommand, shellDirectory);
17.     }
19.     @Override
20.     protected String shellInterpreter() {
21.         return "bash";
22.     }
24.     @Override
25.     protected String shellExtension() {
26.         return ".sh";
27.     }
29.     @Override
30.     protected String shellHeader() {
31.         return "#!/bin/bash";
32.     }
34. }

复制代码

org.apache.dolphinscheduler.plugin.task.api.AbstractCommandExecutor#run

1. public TaskResponse run(IShellInterceptorBuilder iShellInterceptorBuilder,
2.                             TaskCallBack taskCallBack) throws Exception {
3.         TaskResponse result = new TaskResponse();
4.         int taskInstanceId = taskRequest.getTaskInstanceId();
5.         // todo: we need to use state like JDK Thread to make sure the killed task should not be executed
6.         iShellInterceptorBuilder = iShellInterceptorBuilder
7.                 // TODO 设置执行路径
8.                 .shellDirectory(taskRequest.getExecutePath())
9.                 // TODO 这里设置shell 名字
10.                 .shellName(taskRequest.getTaskAppId());
12.         // Set system env
13.         // TODO 在这里是设置默认的，其实也是可以设置为 /opt/dolphinscheduler/bin/env/dolphinscheduler_env.sh
14.         if (CollectionUtils.isNotEmpty(ShellUtils.ENV_SOURCE_LIST)) {
15.             // TODO 这里其实就是向 systemEnvs 中加入ENV_SOURCE_LIST中配置的环境文件的列表
16.             ShellUtils.ENV_SOURCE_LIST.forEach(iShellInterceptorBuilder::appendSystemEnv);
17.         }
19.         // Set custom env
20.         // TODO 设置自定义的env
21.         if (StringUtils.isNotBlank(taskRequest.getEnvironmentConfig())) {
22.             // TODO 向 customEnvScripts 中加入
23.             iShellInterceptorBuilder.appendCustomEnvScript(taskRequest.getEnvironmentConfig());
24.         }
26.         // Set k8s config (This is only work in Linux)
27.         if (taskRequest.getK8sTaskExecutionContext() != null) {
28.             iShellInterceptorBuilder.k8sConfigYaml(taskRequest.getK8sTaskExecutionContext().getConfigYaml());
29.         }
31.         // Set sudo (This is only work in Linux)
32.         // TODO 设置sudo为true的模式
33.         iShellInterceptorBuilder.sudoMode(OSUtils.isSudoEnable());
34.         // Set tenant (This is only work in Linux)
35.         // TODO 设置租户
36.         iShellInterceptorBuilder.runUser(taskRequest.getTenantCode());
37.         // Set CPU Quota (This is only work in Linux)
38.         if (taskRequest.getCpuQuota() != null) {
39.             iShellInterceptorBuilder.cpuQuota(taskRequest.getCpuQuota());
40.         }
41.         // Set memory Quota (This is only work in Linux)
42.         if (taskRequest.getMemoryMax() != null) {
43.             iShellInterceptorBuilder.memoryQuota(taskRequest.getMemoryMax());
44.         }
46.         IShellInterceptor iShellInterceptor = iShellInterceptorBuilder.build();
47.         // TODO 使用ProcessBuilder进行执行，支持sudo模式，和bash模式
48.         process = iShellInterceptor.execute();
50.         // parse process output
51.         // TODO 这里解析到进程的输出
52.         parseProcessOutput(this.process);
54.         // collect pod log
55.         collectPodLogIfNeeded();
57.         int processId = getProcessId(this.process);
59.         result.setProcessId(processId);
61.         // cache processId
62.         taskRequest.setProcessId(processId);
64.         // print process id
65.         log.info("process start, process id is: {}", processId);
67.         // if timeout occurs, exit directly
68.         long remainTime = getRemainTime();
70.         // update pid before waiting for the run to finish
71.         if (null != taskCallBack) {
72.             // TODO 这里其实就是更新任务实例西悉尼
73.             taskCallBack.updateTaskInstanceInfo(taskInstanceId);
74.         }
76.         // waiting for the run to finish
77.         boolean status = this.process.waitFor(remainTime, TimeUnit.SECONDS);
79.         TaskExecutionStatus kubernetesStatus =
80.                 ProcessUtils.getApplicationStatus(taskRequest.getK8sTaskExecutionContext(), taskRequest.getTaskAppId());
82.         if (taskOutputFuture != null) {
83.             try {
84.                 // Wait the task log process finished.
85.                 taskOutputFuture.get();
86.             } catch (ExecutionException e) {
87.                 log.error("Handle task log error", e);
88.             }
89.         }
91.         if (podLogOutputFuture != null) {
92.             try {
93.                 // Wait kubernetes pod log collection finished
94.                 podLogOutputFuture.get();
95.                 // delete pod after successful execution and log collection
96.                 ProcessUtils.cancelApplication(taskRequest);
97.             } catch (ExecutionException e) {
98.                 log.error("Handle pod log error", e);
99.             }
100.         }
102.         // if SHELL task exit
103.         if (status && kubernetesStatus.isSuccess()) {
105.             // SHELL task state
106.             result.setExitStatusCode(this.process.exitValue());
108.         } else {
109.             log.error("process has failure, the task timeout configuration value is:{}, ready to kill ...",
110.                     taskRequest.getTaskTimeout());
111.             result.setExitStatusCode(EXIT_CODE_FAILURE);
112.             cancelApplication();
113.         }
114.         int exitCode = this.process.exitValue();
115.         String exitLogMessage = EXIT_CODE_KILL == exitCode ? "process has killed." : "process has exited.";
116.         log.info("{} execute path:{}, processId:{} ,exitStatusCode:{} ,processWaitForStatus:{} ,processExitValue:{}",
117.                 exitLogMessage, taskRequest.getExecutePath(), processId, result.getExitStatusCode(), status, exitCode);
118.         return result;
120.     }

复制代码

重点就是:

1. // Set system env
2.         // TODO 在这里是设置默认的，其实也是可以设置为 /opt/dolphinscheduler/bin/env/dolphinscheduler_env.sh
3.         if (CollectionUtils.isNotEmpty(ShellUtils.ENV_SOURCE_LIST)) {
4.             // TODO 这里其实就是向 systemEnvs 中加入ENV_SOURCE_LIST中配置的环境文件的列表
5.             ShellUtils.ENV_SOURCE_LIST.forEach(iShellInterceptorBuilder::appendSystemEnv);
6.         }
8.         // Set custom env
9.         // TODO 设置自定义的env
10.         if (StringUtils.isNotBlank(taskRequest.getEnvironmentConfig())) {
11.             // TODO 向 customEnvScripts 中加入
12.             iShellInterceptorBuilder.appendCustomEnvScript(taskRequest.getEnvironmentConfig());
13.         }

复制代码

实在就是说自定的环境变量是可以覆盖默认的环境变量的。
转载自Journey
原文链接：https://segmentfault.com/a/1190000044954252

本文由 白鲸开源 提供发布支持！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。