登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 安全 ›【实战】一招搞定Shell调度！DolphinScheduler+ProcessB ...

【实战】一招搞定Shell调度！DolphinScheduler+ProcessBuilder超详细教程

訾懵 6 天前
本文将介绍在DolphinScheduler中使用ProcessBuilder执行Shell命令的方法。默认通过BashShellInterceptorBuilder封装Shell脚本并生成执行命令，支持普通模式和sudo模式运行。同时，结合Spring Boot应用示例，展示了如何配置工作目录、合并错误流、监控执行状态，并输出日志信息，从而实现对Shell任务的高效管理和调度。
1、ProcessBuilder DolphinScheduler中的使用

1.1、命令的封装

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.     }
  19. }
复制代码
默认走的是 BashShellInterceptorBuilder。
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.     // 这个是如果不是sudo的方式，进行命令执行的前缀
  20.     @Override
  21.     protected String shellInterpreter() {
  22.         return "bash";
  23.     }
  25.     @Override
  26.     protected String shellExtension() {
  27.         return ".sh";
  28.     }
  30.     @Override
  31.     protected String shellHeader() {
  32.         return "#!/bin/bash";
  33.     }
  34. }
复制代码
org.apache.dolphinscheduler.plugin.task.api.shell.BaseLinuxShellInterceptorBuilder#generateBootstrapCommand
  1. protected List<String> generateBootstrapCommand() {
  2.         if (sudoEnable) {
  3.             // TODO 默认是走这里的，其实就是sudo -u 租户 -i /opt/xx.sh
  4.             return bootstrapCommandInSudoMode();
  5.         }
  6.         // TODO bash /opt/xx.sh
  7.         return bootstrapCommandInNormalMode();
  8.     }
复制代码
bootstrapCommandInSudoMode():
  1. private List<String>
  4. bootstrapCommandInSudoMode() {
  5.         if (PropertyUtils.getBoolean(AbstractCommandExecutorConstants.TASK_RESOURCE_LIMIT_STATE, false)) {
  6.             return bootstrapCommandInResourceLimitMode();
  7.         }
  8.         List<String> bootstrapCommand = new ArrayList<>();
  9.         bootstrapCommand.add("sudo");
  10.         if (StringUtils.isNotBlank(runUser)) {
  11.             bootstrapCommand.add("-u");
  12.             bootstrapCommand.add(runUser);
  13.         }
  14.         bootstrapCommand.add("-i");
  15.         bootstrapCommand.add(shellAbsolutePath().toString());
  16.         return bootstrapCommand;
  17.     }
复制代码
bootstrapCommandInNormalMode():
  1. private List<String> bootstrapCommandInNormalMode() {
  2.         List<String> bootstrapCommand = new ArrayList<>();
  3.         bootstrapCommand.add(shellInterpreter());
  4.         bootstrapCommand.add(shellAbsolutePath().toString());
  5.         return bootstrapCommand;
  6.     }
复制代码
1.2、命令的执行

org.apache.dolphinscheduler.plugin.task.api.shell.BaseShellInterceptor
  1. public abstract class BaseShellInterceptor implements IShellInterceptor {
  3.     protected final String workingDirectory;
  4.     protected final List<String> executeCommands;
  6.     protected BaseShellInterceptor(List<String> executeCommands, String workingDirectory) {
  7.         this.executeCommands = executeCommands;
  8.         this.workingDirectory = workingDirectory;
  9.     }
  11.     @Override
  12.     public Process execute() throws IOException {
  13.         // init process builder
  14.         ProcessBuilder processBuilder = new ProcessBuilder();
  15.         // setting up a working directory
  16.         // TODO 设置工作路径，目的其实就是在执行脚本的时候，可以在该目录的位置来加载比如说jar包什么的
  17.         processBuilder.directory(new File(workingDirectory));
  18.         // merge error information to standard output stream
  19.         processBuilder.redirectErrorStream(true);
  20.         processBuilder.command(executeCommands);
  21.         log.info("Executing shell command : {}", String.join(" ", executeCommands));
  22.         return processBuilder.start();
  23.     }
  24. }
复制代码
2、最佳实践实例

2.1、pom.xml配置
  1. <dependency>
  2.   <groupId>org.springframework.boot</groupId>
  3.   spring-boot-starter</artifactId>
  4.   <version>2.6.1</version>
  5. </dependency>
复制代码
2.2、pom.xml配置
  1. @SpringBootApplication
  2. public class Application {
  4.     public static void main(String[] args) throws Exception {
  5.         SpringApplication.run(Application.class, args);
  7.         List<String> executeCommands = new ArrayList<>();
  8.         executeCommands.add("sudo");
  9.         executeCommands.add("-u");
  10.         executeCommands.add("qiaozhanwei");
  11.         executeCommands.add("-i");
  12.         executeCommands.add("/opt/test/my.sh");
  15.         ProcessBuilder processBuilder = new ProcessBuilder();
  16.         // setting up a working directory
  17.         // TODO 设置工作路径，目的其实就是在执行脚本的时候，可以在该目录的位置来加载比如说jar包什么的
  18.         processBuilder.directory(new File("/opt/test"));
  19.         // merge error information to standard output stream
  20.         processBuilder.redirectErrorStream(true);
  21.         processBuilder.command(executeCommands);
  22.         Process process = processBuilder.start();
  24.         try (BufferedReader inReader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
  25.             String line;
  26.             while ((line = inReader.readLine()) != null) {
  27.                 // TODO 终端日志输出
  28.                 System.out.println(line);
  29.             }
  30.         } catch (Exception e) {
  31.             e.printStackTrace();
  32.         }
  35.         // TODO 等10分钟，如果10分钟不结束，返回且status为false
  36.         boolean status = process.waitFor(10, TimeUnit.MINUTES);
  38.         System.out.println("status ->" + status);
  39.     }
  40. }
复制代码
2.3、日志输出结果
  1.   .   ____          _            __ _ _
  2. /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
  3. ( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
  4. \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  5.   '  |____| .__|_| |_|_| |_\__, | / / / /
  6. =========|_|==============|___/=/_/_/_/
  7. :: Spring Boot ::                (v2.6.1)
  9. 2024-06-15 18:33:16.090  INFO 31834 --- [           main] com.journey.test.Application             : Starting Application using Java 1.8.0_401 on 192.168.1.4 with PID 31834 (/Users/qiaozhanwei/IdeaProjects/springboot2/target/classes started by qiaozhanwei in /Users/qiaozhanwei/IdeaProjects/springboot2)
  10. 2024-06-15 18:33:16.091  INFO 31834 --- [           main] com.journey.test.Application             : No active profile set, falling back to default profiles: default
  11. 2024-06-15 18:33:16.244  INFO 31834 --- [           main] com.journey.test.Application             : Started Application in 0.252 seconds (JVM running for 0.42)
  12. Number of Maps  = 1
  13. Samples per Map = 100000
  14. 2024-06-15 18:33:16,790 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
  15. Wrote input for Map #0
  16. Starting Job
  17. 2024-06-15 18:33:17,329 INFO client.DefaultNoHARMFailoverProxyProvider: Connecting to ResourceManager at kvm-10-253-26-85/10.253.26.85:8032
  18. 2024-06-15 18:33:17,586 INFO mapreduce.JobResourceUploader: Disabling Erasure Coding for path: /tmp/hadoop-yarn/staging/qiaozhanwei/.staging/job_1694766249884_0931
  19. 2024-06-15 18:33:17,837 INFO input.FileInputFormat: Total input files to process : 1
  20. 2024-06-15 18:33:18,024 INFO mapreduce.JobSubmitter: number of splits:1
  21. 2024-06-15 18:33:18,460 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1694766249884_0931
  22. 2024-06-15 18:33:18,460 INFO mapreduce.JobSubmitter: Executing with tokens: []
  23. 2024-06-15 18:33:18,648 INFO conf.Configuration: resource-types.xml not found
  24. 2024-06-15 18:33:18,648 INFO resource.ResourceUtils: Unable to find 'resource-types.xml'.
  25. 2024-06-15 18:33:18,698 INFO impl.YarnClientImpl: Submitted application application_1694766249884_0931
  26. 2024-06-15 18:33:18,734 INFO mapreduce.Job: The url to track the job: http://kvm-10-253-26-85:8088/proxy/application_1694766249884_0931/
  27. 2024-06-15 18:33:18,734 INFO mapreduce.Job: Running job: job_1694766249884_0931
  28. 2024-06-15 18:33:24,978 INFO mapreduce.Job: Job job_1694766249884_0931 running in uber mode : false
  29. 2024-06-15 18:33:24,978 INFO mapreduce.Job:  map 0% reduce 0%
  30. 2024-06-15 18:33:29,153 INFO mapreduce.Job:  map 100% reduce 0%
  31. 2024-06-15 18:33:34,384 INFO mapreduce.Job:  map 100% reduce 100%
  32. 2024-06-15 18:33:34,455 INFO mapreduce.Job: Job job_1694766249884_0931 completed successfully
  33. 2024-06-15 18:33:34,565 INFO mapreduce.Job: Counters: 54
  34.     File System Counters
  35.         FILE: Number of bytes read=28
  36.         FILE: Number of bytes written=548863
  37.         FILE: Number of read operations=0
  38.         FILE: Number of large read operations=0
  39.         FILE: Number of write operations=0
  40.         HDFS: Number of bytes read=278
  41.         HDFS: Number of bytes written=215
  42.         HDFS: Number of read operations=9
  43.         HDFS: Number of large read operations=0
  44.         HDFS: Number of write operations=3
  45.         HDFS: Number of bytes read erasure-coded=0
  46.     Job Counters
  47.         Launched map tasks=1
  48.         Launched reduce tasks=1
  49.         Data-local map tasks=1
  50.         Total time spent by all maps in occupied slots (ms)=37968
  51.         Total time spent by all reduces in occupied slots (ms)=79360
  52.         Total time spent by all map tasks (ms)=2373
  53.         Total time spent by all reduce tasks (ms)=2480
  54.         Total vcore-milliseconds taken by all map tasks=2373
  55.         Total vcore-milliseconds taken by all reduce tasks=2480
  56.         Total megabyte-milliseconds taken by all map tasks=4859904
  57.         Total megabyte-milliseconds taken by all reduce tasks=10158080
  58.     Map-Reduce Framework
  59.         Map input records=1
  60.         Map output records=2
  61.         Map output bytes=18
  62.         Map output materialized bytes=28
  63.         Input split bytes=160
  64.         Combine input records=0
  65.         Combine output records=0
  66.         Reduce input groups=2
  67.         Reduce shuffle bytes=28
  68.         Reduce input records=2
  69.         Reduce output records=0
  70.         Spilled Records=4
  71.         Shuffled Maps =1
  72.         Failed Shuffles=0
  73.         Merged Map outputs=1
  74.         GC time elapsed (ms)=87
  75.         CPU time spent (ms)=1420
  76.         Physical memory (bytes) snapshot=870387712
  77.         Virtual memory (bytes) snapshot=9336647680
  78.         Total committed heap usage (bytes)=2716860416
  79.         Peak Map Physical memory (bytes)=457416704
  80.         Peak Map Virtual memory (bytes)=3773362176
  81.         Peak Reduce Physical memory (bytes)=412971008
  82.         Peak Reduce Virtual memory (bytes)=5563285504
  83.     Shuffle Errors
  84.         BAD_ID=0
  85.         CONNECTION=0
  86.         IO_ERROR=0
  87.         WRONG_LENGTH=0
  88.         WRONG_MAP=0
  89.         WRONG_REDUCE=0
  90.     File Input Format Counters
  91.         Bytes Read=118
  92.     File Output Format Counters
  93.         Bytes Written=97
  94. Job Finished in 17.292 seconds
  95. Estimated value of Pi is 3.14120000000000000000
  96. status ->true
  98. Process finished with exit code 0
复制代码
转载自Journey
原文链接：https://segmentfault.com/a/1190000044966157
本文由 白鲸开源 提供发布支持！

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
訾懵
6 天前

0

粉丝关注

12

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫