ToB企服应用市场:ToB评测及商务社交产业平台

标题: 使用java远程提交flink任务到yarn集群 [打印本页]
作者: 缠丝猫    时间: 2024-7-27 00:38
标题: 使用java远程提交flink任务到yarn集群
使用java远程提交flink任务到yarn集群

背景

由于业务需要，使用命令行的方式提交flink任务比较麻烦，要么将后端任务部署到大数据集群，要么弄一个提交机，感觉都不是很离线。颠末一些调研，发现可以实现远程的任务发布。接下来就记录一下实现过程。这里用flink on yarn 的Application模式实现
情况预备


正式开始

1. 上传flink jar包到hdfs

去flink官网下载你需要的版本，我这里用的是flink-1.18.1，把flink lib目次下的jar包传到hdfs中。

其中flink-yarn-1.18.1.jar需要各人本身去maven仓库下载。
2. 编写一段flink代码

随便写一段flink代码就行，我们目的是测试
  1. package com.azt;
  3. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  7. import java.util.Random;
  8. import java.util.concurrent.TimeUnit;
  10. public class WordCount {
  11.     public static void main(String[] args) throws Exception {
  12.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  13.         DataStreamSource<String> source = env.addSource(new SourceFunction<String>() {
  14.             @Override
  15.             public void run(SourceContext<String> ctx) throws Exception {
  16.                 String[] words = {"spark", "flink", "hadoop", "hdfs", "yarn"};
  17.                 Random random = new Random();
  18.                 while (true) {
  19.                     ctx.collect(words[random.nextInt(words.length)]);
  20.                     TimeUnit.SECONDS.sleep(1);
  21.                 }
  22.             }
  23.             @Override
  24.             public void cancel() {
  26.             }
  27.         });
  29.         source.print();
  30.         env.execute();
  31.     }
  32. }
复制代码
3. 打包第二步的代码，上传到hdfs


4. 拷贝配置文件


具体看截图

5. 编写java远程提交任务的程序

这一步有个留意的地方就是，如果你跟我一样是windows电脑，那么本地用idea提交会报错；如果你是mac或者linux，那么可以直接在idea中提交任务。
  1. package com.test;
  4. import org.apache.flink.client.deployment.ClusterDeploymentException;
  5. import org.apache.flink.client.deployment.ClusterSpecification;
  6. import org.apache.flink.client.deployment.application.ApplicationConfiguration;
  7. import org.apache.flink.client.program.ClusterClient;
  8. import org.apache.flink.client.program.ClusterClientProvider;
  9. import org.apache.flink.configuration.*;
  10. import org.apache.flink.runtime.client.JobStatusMessage;
  11. import org.apache.flink.yarn.YarnClientYarnClusterInformationRetriever;
  12. import org.apache.flink.yarn.YarnClusterDescriptor;
  13. import org.apache.flink.yarn.YarnClusterInformationRetriever;
  14. import org.apache.flink.yarn.configuration.YarnConfigOptions;
  15. import org.apache.flink.yarn.configuration.YarnDeploymentTarget;
  17. import org.apache.flink.yarn.configuration.YarnLogConfigUtil;
  18. import org.apache.hadoop.fs.Path;
  19. import org.apache.hadoop.yarn.api.records.ApplicationId;
  20. import org.apache.hadoop.yarn.client.api.YarnClient;
  21. import org.apache.hadoop.yarn.conf.YarnConfiguration;
  23. import java.util.ArrayList;
  24. import java.util.Collection;
  25. import java.util.Collections;
  26. import java.util.List;
  27. import java.util.concurrent.CompletableFuture;
  29. import static org.apache.flink.configuration.MemorySize.MemoryUnit.MEGA_BYTES;
  31. /**
  32. * @date ：2021/5/12 7:16 下午
  33. */
  34. public class Main {
  35.     public static void main(String[] args) throws Exception {
  36.         ///home/root/flink/lib/lib
  37.         System.setProperty("HADOOP_USER_NAME","root");
  38. //        String configurationDirectory = "C:\\project\\test_flink_mode\\src\\main\\resources\\conf";
  39.         String configurationDirectory = "/export/server/flink-1.18.1/conf";
  40.         org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
  41.         conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");
  42.         conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
  43.         String flinkLibs = "hdfs://node1.itcast.cn/flink/lib";
  44.         String userJarPath = "hdfs://node1.itcast.cn/flink/user-lib/original.jar";
  45.         String flinkDistJar = "hdfs://node1.itcast.cn/flink/lib/flink-yarn-1.18.1.jar";
  47.         YarnClient yarnClient = YarnClient.createYarnClient();
  48.         YarnConfiguration yarnConfiguration = new YarnConfiguration();
  49.         yarnClient.init(yarnConfiguration);
  50.         yarnClient.start();
  52.         YarnClusterInformationRetriever clusterInformationRetriever = YarnClientYarnClusterInformationRetriever
  53.                 .create(yarnClient);
  54.         //获取flink的配置
  55.         Configuration flinkConfiguration = GlobalConfiguration.loadConfiguration(
  56.                 configurationDirectory);
  57.         flinkConfiguration.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
  58.         flinkConfiguration.set(
  59.                 PipelineOptions.JARS,
  60.                 Collections.singletonList(
  61.                         userJarPath));
  62.         YarnLogConfigUtil.setLogConfigFileInConfig(flinkConfiguration,configurationDirectory);
  63.         Path remoteLib = new Path(flinkLibs);
  64.         flinkConfiguration.set(
  65.                 YarnConfigOptions.PROVIDED_LIB_DIRS,
  66.                 Collections.singletonList(remoteLib.toString()));
  68.         flinkConfiguration.set(
  69.                 YarnConfigOptions.FLINK_DIST_JAR,
  70.                 flinkDistJar);
  71.         //设置为application模式
  72.         flinkConfiguration.set(
  73.                 DeploymentOptions.TARGET,
  74.                 YarnDeploymentTarget.APPLICATION.getName());
  75.         //yarn application name
  76.         flinkConfiguration.set(YarnConfigOptions.APPLICATION_NAME, "jobname");
  77.         //设置配置，可以设置很多
  78.         flinkConfiguration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1024",MEGA_BYTES));
  79.         flinkConfiguration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1024",MEGA_BYTES));
  80.         flinkConfiguration.set(TaskManagerOptions.NUM_TASK_SLOTS, 4);
  81.         flinkConfiguration.setInteger("parallelism.default", 4);
  83.         ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
  84.                 .createClusterSpecification();
  86. //                设置用户jar的参数和主类
  87.         ApplicationConfiguration appConfig = new ApplicationConfiguration(args,"com.azt.WordCount");
  90.         YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(
  91.                 flinkConfiguration,
  92.                 yarnConfiguration,
  93.                 yarnClient,
  94.                 clusterInformationRetriever,
  95.                 true);
  96.         ClusterClientProvider<ApplicationId> clusterClientProvider = null;
  97.         try {
  98.             clusterClientProvider = yarnClusterDescriptor.deployApplicationCluster(
  99.                     clusterSpecification,
  100.                     appConfig);
  101.         } catch (ClusterDeploymentException e){
  102.             e.printStackTrace();
  103.         }
  105.         ClusterClient<ApplicationId> clusterClient = clusterClientProvider.getClusterClient();
  106.         System.out.println(clusterClient.getWebInterfaceURL());
  107.         ApplicationId applicationId = clusterClient.getClusterId();
  109.         System.out.println(applicationId);
  111.         Collection<JobStatusMessage> jobStatusMessages = clusterClient.listJobs().get();
  112.         int counts = 30;
  113.         while (jobStatusMessages.size() == 0 && counts > 0) {
  114.             Thread.sleep(1000);
  115.             counts--;
  116.             jobStatusMessages = clusterClient.listJobs().get();
  117.             if (jobStatusMessages.size() > 0) {
  118.                 break;
  119.             }
  120.         }
  121.         if (jobStatusMessages.size() > 0) {
  122.             List<String> jids = new ArrayList<>();
  123.             for (JobStatusMessage jobStatusMessage : jobStatusMessages) {
  124.                 jids.add(jobStatusMessage.getJobId().toHexString());
  125.             }
  126.             System.out.println(String.join(",",jids));
  127.         }
  129.     }
  130. }
复制代码
由于我这里是windows电脑，以是我打包放到服务器上去运行
执行命令 ：
   java -cp test_flink_mode-1.0-SNAPSHOT.jar com.test.Main
  不出以外的话，会打印如下日志
  1. log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).
  2. log4j:WARN Please initialize the log4j system properly.
  3. log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
  4. http://node2:33811
  5. application_1715418089838_0017
  6. 6d4d6ed5277a62fc9a3a274c4f34a468
复制代码
复制打印的url毗连，就可以打开flink的webui了，在yarn的前端页面中也可以看到flink任务。

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



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4