使用java远程提交flink任务到yarn集群

使用java远程提交flink任务到yarn集群

背景

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

环境准备

  • 大数据集群，只要有hadoop就行
  • 后端服务器，linux mac都行，windows不行

正式开始

1. 上传flink jar包到hdfs

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

在这里插入图片描述
其中flink-yarn-1.18.1.jar需要大家自己去maven仓库下载。

2. 编写一段flink代码

随便写一段flink代码就行，我们目的是测试

package com.azt;import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;import java.util.Random;
import java.util.concurrent.TimeUnit;public class WordCount {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> source = env.addSource(new SourceFunction<String>() {@Overridepublic void run(SourceContext<String> ctx) throws Exception {String[] words = {"spark", "flink", "hadoop", "hdfs", "yarn"};Random random = new Random();while (true) {ctx.collect(words[random.nextInt(words.length)]);TimeUnit.SECONDS.sleep(1);}}@Overridepublic void cancel() {}});source.print();env.execute();}
}

3. 打包第二步的代码，上传到hdfs

在这里插入图片描述

4. 拷贝配置文件

  • 拷贝flink conf下的所有文件到java项目的resource中
  • 拷贝hadoop配置文件到到java项目的resource中

具体看截图
在这里插入图片描述

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

这一步有个注意的地方就是，如果你跟我一样是windows电脑，那么本地用idea提交会报错；如果你是mac或者linux，那么可以直接在idea中提交任务。

package com.test;import org.apache.flink.client.deployment.ClusterDeploymentException;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.*;
import org.apache.flink.runtime.client.JobStatusMessage;
import org.apache.flink.yarn.YarnClientYarnClusterInformationRetriever;
import org.apache.flink.yarn.YarnClusterDescriptor;
import org.apache.flink.yarn.YarnClusterInformationRetriever;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.flink.yarn.configuration.YarnDeploymentTarget;import org.apache.flink.yarn.configuration.YarnLogConfigUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;import static org.apache.flink.configuration.MemorySize.MemoryUnit.MEGA_BYTES;/*** @date ：2021/5/12 7:16 下午*/
public class Main {public static void main(String[] args) throws Exception {///home/root/flink/lib/libSystem.setProperty("HADOOP_USER_NAME","root");
//        String configurationDirectory = "C:\\project\\test_flink_mode\\src\\main\\resources\\conf";String configurationDirectory = "/export/server/flink-1.18.1/conf";org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());String flinkLibs = "hdfs://node1.itcast.cn/flink/lib";String userJarPath = "hdfs://node1.itcast.cn/flink/user-lib/original.jar";String flinkDistJar = "hdfs://node1.itcast.cn/flink/lib/flink-yarn-1.18.1.jar";YarnClient yarnClient = YarnClient.createYarnClient();YarnConfiguration yarnConfiguration = new YarnConfiguration();yarnClient.init(yarnConfiguration);yarnClient.start();YarnClusterInformationRetriever clusterInformationRetriever = YarnClientYarnClusterInformationRetriever.create(yarnClient);//获取flink的配置Configuration flinkConfiguration = GlobalConfiguration.loadConfiguration(configurationDirectory);flinkConfiguration.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);flinkConfiguration.set(PipelineOptions.JARS,Collections.singletonList(userJarPath));YarnLogConfigUtil.setLogConfigFileInConfig(flinkConfiguration,configurationDirectory);Path remoteLib = new Path(flinkLibs);flinkConfiguration.set(YarnConfigOptions.PROVIDED_LIB_DIRS,Collections.singletonList(remoteLib.toString()));flinkConfiguration.set(YarnConfigOptions.FLINK_DIST_JAR,flinkDistJar);//设置为application模式flinkConfiguration.set(DeploymentOptions.TARGET,YarnDeploymentTarget.APPLICATION.getName());//yarn application nameflinkConfiguration.set(YarnConfigOptions.APPLICATION_NAME, "jobname");//设置配置，可以设置很多flinkConfiguration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1024",MEGA_BYTES));flinkConfiguration.set(TaskManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("1024",MEGA_BYTES));flinkConfiguration.set(TaskManagerOptions.NUM_TASK_SLOTS, 4);flinkConfiguration.setInteger("parallelism.default", 4);ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();//		设置用户jar的参数和主类ApplicationConfiguration appConfig = new ApplicationConfiguration(args,"com.azt.WordCount");YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(flinkConfiguration,yarnConfiguration,yarnClient,clusterInformationRetriever,true);ClusterClientProvider<ApplicationId> clusterClientProvider = null;try {clusterClientProvider = yarnClusterDescriptor.deployApplicationCluster(clusterSpecification,appConfig);} catch (ClusterDeploymentException e){e.printStackTrace();}ClusterClient<ApplicationId> clusterClient = clusterClientProvider.getClusterClient();System.out.println(clusterClient.getWebInterfaceURL());ApplicationId applicationId = clusterClient.getClusterId();System.out.println(applicationId);Collection<JobStatusMessage> jobStatusMessages = clusterClient.listJobs().get();int counts = 30;while (jobStatusMessages.size() == 0 && counts > 0) {Thread.sleep(1000);counts--;jobStatusMessages = clusterClient.listJobs().get();if (jobStatusMessages.size() > 0) {break;}}if (jobStatusMessages.size() > 0) {List<String> jids = new ArrayList<>();for (JobStatusMessage jobStatusMessage : jobStatusMessages) {jids.add(jobStatusMessage.getJobId().toHexString());}System.out.println(String.join(",",jids));}}
}

由于我这里是windows电脑，所以我打包放到服务器上去运行
执行命令 ：

java -cp test_flink_mode-1.0-SNAPSHOT.jar com.test.Main

不出以外的话，会打印如下日志

log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
http://node2:33811
application_1715418089838_0017
6d4d6ed5277a62fc9a3a274c4f34a468

复制打印的url连接，就可以打开flink的webui了，在yarn的前端页面中也可以看到flink任务。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/696273.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

网络基础-SSH协议（思科、华为、华三）

网络基础-SSH协议（思科、华为、华三）

SSH&#xff08;Secure Shell&#xff09;是一种用于安全远程访问和安全文件传输的协议。它提供了加密的通信通道&#xff0c;使得用户可以在不安全的网络上安全地远程登录到远程主机&#xff0c;并在远程主机上执行命令、访问文件以及传输文件&#xff0c;本篇主要讲解命令执行…
阅读更多...
M 有效算法

M 有效算法

M 有效算法 本题考验二分知识&#xff0c;思路是二分k的取值&#xff0c;就按第一组样例来说当我们k取值为1的时候我们遍历数组想让|8-x|<k1的话x的取值范围是7-9&#xff0c;想让|3-x|<k2的话x的取值范围是1-5&#xff0c;两者x的区间不重合&#xff0c;说明肯定没有x能…
阅读更多...
Linux 中 alarm 函数详解

Linux 中 alarm 函数详解

目录 简介函数原型函数参数返回值使用示例设置 3 秒闹钟修改闹钟与取消闹钟设置 1 秒周期定时器 更多内容 简介 alarm 函数的功能是设置一个闹钟&#xff08;定时器&#xff09;&#xff0c;当闹钟时间到时&#xff0c;内核会向当前进程发送一个 SIGALRM 信号。 打开 Linux 终…
阅读更多...
网络编程：服务器模型-并发服务器-多线程实现并发服务器

网络编程：服务器模型-并发服务器-多线程实现并发服务器

思想&#xff1a; 主进程专门用于连接多个客户端的请求&#xff0c;若有一条客户端连接进来&#xff0c;主进程就创建一个子进程&#xff0c;用该子进程来处理客户端的业务数据。 tcp_server.c #include <sys/types.h> #include <sys/socket.h> #include <std…
阅读更多...
2024期从、证从、基从、银从备考资料及互助交流群

2024期从、证从、基从、银从备考资料及互助交流群

快进&#xff0c;2024年 金融考证备考资料及互助交流备考群&#xff0c;考啥进啥 5.12 &#xff08;基从&#xff09;基金从业资格考试 5.18&#xff08;期从&#xff09;期货从业资格考试 6.1 &#xff08;证从&#xff09;证券从业资格考试 6.1&#xff5e;6.2&#xff08;银…
阅读更多...
交通地理信息系统实习教程（二）

交通地理信息系统实习教程（二）

这篇文章服务于GIS背景用户以及有志于GIS的朋友 操作源数据位置&#xff1a;【免费】交通地理信息系统实习二源数据资源-CSDN文库 软件安装包位置&#xff1a;【免费】TransCad-交通地理信息系统软件资源-CSDN文库 一、最短路径分析 1.1软件启动说明 这里需要给出一个必要的…
阅读更多...
福建医疗器械展/2024厦门国际医疗器械展览会重磅来袭

福建医疗器械展/2024厦门国际医疗器械展览会重磅来袭

2024中国&#xff08;厦门&#xff09;国际医疗器械展览会 时 间&#xff1a;2024年11月1-3日 November 1-3, 2024 地 点&#xff1a;厦门国际会展中心 Xiamen International Conference & Exhibition Center ​ ◆组织机构 主办单位&#xff1a; 中国技术市场协会医…
阅读更多...
Java——多线程

Java——多线程

一.多线程 1.什么是多线程 线程是操作系统能够进行运算调度的最小单位。它被包含在进程之中&#xff0c;是进程的实际运作单位 简单理解多线程就是应用软件中相互独立&#xff0c;可以同时运行的功能(也可以理解为人体内相互独立&#xff0c;但可以同时运行的器官⌓‿⌓) 我们…
阅读更多...
代理IP可靠吗？哪里可以找到可靠的代理？

代理IP可靠吗？哪里可以找到可靠的代理？

需要代理来访问受限制的网站或改善您的在线隐私&#xff1f;别再犹豫了&#xff01;在这篇博文中&#xff0c;我们将探讨您可以使用的选项&#xff0c;并提供有关在哪里获取代理的指导。 首先&#xff0c;让我们了解什么是代理及其工作原理。代理充当您的设备和互联网之间的中介…
阅读更多...
ATFX：美国通胀率平台期，或助力黄金延续涨势

ATFX：美国通胀率平台期，或助力黄金延续涨势

ATFX金属&#xff1a;5月9日19:00至5月10日19:00&#xff0c;COMEX黄金的小时级别出现一波持续24小时的上涨走势&#xff0c;最高触及2385.3美元&#xff0c;累计涨幅2.78%&#xff0c;成为上周最佳的短线交易时机。R阻力线形成后&#xff0c;COMEX黄金进入下降通道&#xff0c…
阅读更多...
MySQL从入门到高级 --- 6.函数

MySQL从入门到高级 --- 6.函数

文章目录 第六章&#xff1a;6.函数6.1 聚合函数6.2 数学函数6.3 字符串函数6.4 日期函数6.4.1 日期格式 6.5 控制流函数6.5.1 if逻辑判断语句6.5.2 case when语句 6.6 窗口函数6.6.1 序号函数6.6.2 开窗聚合函数6.6.3 分布函数6.6.4 前后函数6.6.5 头尾函数6.6.6 其他函数6.7 …
阅读更多...
Golang面向对象编程（二）

Golang面向对象编程（二）

文章目录 封装基本介绍封装的实现工厂函数 继承基本介绍继承的实现字段和方法访问细节多继承 封装 基本介绍 基本介绍 封装&#xff08;Encapsulation&#xff09;是面向对象编程&#xff08;OOP&#xff09;中的一种重要概念&#xff0c;封装通过将数据和相关的方法组合在一起…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023