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

news2024/11/27 18:42:02

使用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>() {
            @Override
            public 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);
                }
            }
            @Override
            public 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/lib
        System.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 name
        flinkConfiguration.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.coloradmin.cn/o/1671026.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【RabbitMQ】消息队列 - RabbitMQ的使用记录

【RabbitMQ】消息队列 - RabbitMQ的使用记录

目录 一、什么是消息队列 二、什么是RabbitMQ 三、安装RabbitMQ 3.1 安装Erlang环境 3.2 安装RabbitMQ 3.3 打开服务管理界面 3.4 常用命令 四、Python示例代码 4.1 发送数据 4.2 接收数据 一、什么是消息队列 消息队列(Message Queue)是一种用于在应用程序之间传递消…
阅读更多...
vue：网页icon无法显示

vue：网页icon无法显示

logo文件放在public文件夹下&#xff0c;在html里设置icon。 本地源码运行后发现网页icon无法显示我们设置的logo&#xff0c;而是显示了浏览器默认icon。 这个问题不需要解决&#xff0c;部署后网页icon显示就正常了。
阅读更多...
绝地求生PUBG新老艾伦格有什么差别 老艾伦格什么时候回归

绝地求生PUBG新老艾伦格有什么差别 老艾伦格什么时候回归

复古风格的艾伦格原始地图携带着那些标志性的记忆符号华丽回归&#xff0c;邀请您沉浸于往昔的每一处细节探索中。我们不仅还原了游戏诞生的起点&#xff0c;还在其中巧妙融入现代游戏元素&#xff0c;构筑一座连接昔日与今朝的桥梁&#xff0c;完美融合了经典与创新的游戏体验…
阅读更多...
【动态规划四】子序列问题

【动态规划四】子序列问题

目录 leetcode题目 一、最长递增子序列 二、摆动序列 三、最长递增子序列的个数 四、最长数对链 五、最长定差子序列 六、最长的斐波那契子序列的长度 七、最长等差数列 八、等差数列划分 II leetcode题目 一、最长递增子序列 300. 最长递增子序列 - 力扣&#xff0…
阅读更多...
关于SQL

关于SQL

数据库简介&#xff1a; 数据库分类 关系型数据库模型&#xff1a; 优点&#xff1a;易于维护&#xff0c;可以实现复杂的查询 缺点&#xff1a;海量数据 读取写入性能差&#xff0c;高并发下数据库的io是瓶颈 是把复杂的数据结构归结为简单的二元关系&#xff08;即二维表…
阅读更多...
Linux字符设备驱动设计

Linux字符设备驱动设计

Linux字符设备驱动设计 概述 驱动的定义与功能 计算机系统中存在着大量的设备&#xff0c; 操作系统要求能够控制和管理这些硬件&#xff0c; 而驱动就是帮助操作系统完成这个任务。 驱动相当于硬件的接口&#xff0c; 它直接操作、 控制着我们的硬件&#xff0c; 操作系统通…
阅读更多...
工作太闲，平常有没有用手机能赚钱的，这里我推荐了4种可月入2000的副业

工作太闲，平常有没有用手机能赚钱的，这里我推荐了4种可月入2000的副业

确实有许多通过手机赚钱的方式&#xff0c;以下是一些常见的方法 1.拍照卖图 如果你有一台相机或智能手机&#xff0c;喜欢拍照&#xff0c;那么可以将自己拍摄的图片上传到网站上&#xff0c;赚取稿费。 2. 做问卷调查 许多公司需要了解消费者的意见&#xff0c;所以会通过…
阅读更多...
【XSRP软件无线电】基于软件无线电平台的QPSK频带通信系统设计

【XSRP软件无线电】基于软件无线电平台的QPSK频带通信系统设计

目录&#xff1a; 目录&#xff1a; 一、绪论 1.1 设计背景 1.2 设计目的 二、系统总体方案 2.1 专题调研题目 2.2 调研背景 2.3 设计任务解读 2.4 设计原理 2.4.1 原理框图 2.4.2 功能验证 三、软件设计 3.1 程序解读 3.2 程序设计 3.3 仿真结果&#xff1a; 四、程序代码分析…
阅读更多...
爬取深圳2024年链家二手房数据，共3000条数据（其他城市也可）

爬取深圳2024年链家二手房数据，共3000条数据（其他城市也可）

文章目录 专栏导读1.目标2.导入相关库3.获取每个二手房的链接4.获取每个链接中的相关数据5.保存数据6.数据展示 专栏导读 ✍ 作者简介&#xff1a;i阿极&#xff0c;CSDN 数据分析领域优质创作者&#xff0c;专注于分享python数据分析领域知识。 ✍ 本文录入于《python网络爬虫…
阅读更多...
二叉树OJ刷题

二叉树OJ刷题

制作不易&#xff0c;三连支持一下吧&#xff01;&#xff01;&#xff01; 文章目录 前言一、相同的树二、单值二叉树三.对称二叉树四.二叉树的前序遍历五.另一棵树的子树六.二叉树遍历总结 前言 前三篇博客我们详细介绍了树形结构&#xff0c;及两种特殊的树&#xff1a;堆和…
阅读更多...
【Ubuntu永久授权串口设备读取权限“/dev/ttyUSB0”】

【Ubuntu永久授权串口设备读取权限“/dev/ttyUSB0”】

Ubuntu永久授权串口设备读取权限 1 问题描述2 解决方案2.1 查看ttyUSB0权限&#xff0c;拥有者是root&#xff0c;所属用户组为dialout2.2 查看dialout用户组成员&#xff0c;如图所示&#xff0c;普通用户y不在dialout组中2.3 将普通用户y加入dialout组中2.4 再次查看dialout用…
阅读更多...
云原生新手和开源教育分论坛 02-技术 or 非技术，参与 Kubernetes 社区丝滑路径【开源贡献】

云原生新手和开源教育分论坛 02-技术 or 非技术，参与 Kubernetes 社区丝滑路径【开源贡献】

https://www.kubernetes.dev/https://www.kubernetes.dev/community/community-groups/https://killercoda.com/https://kwok.sigs.k8s.io/https://training.linuxfoundation.cn/ 演讲
阅读更多...
pywinauto，一款Win自动化利器！

pywinauto，一款Win自动化利器！

pywinauto&#xff0c;一款Win自动化利器&#xff01; 1.安装 pywinauto是一个用于自动化Python模块&#xff0c;适合Windows系统的软件&#xff08;GUI&#xff09;&#xff0c;可以通过Pywinauto遍历窗口&#xff08;对话框&#xff09;和窗口里的控件&#xff0c;也可以控…
阅读更多...
NeurIPS‘24 截稿日期逼近 加拿大温哥华邀你共赴盛会

NeurIPS‘24 截稿日期逼近 加拿大温哥华邀你共赴盛会

会议之眼 快讯 第38届NeurIPS24(Conference and Workshop on Neural Information Processing Systems)即神经信息处理系统研讨会将于 2024 年 12月9日-15日在加拿大温哥华会议中心举行&#xff01; NeurIPS 每一年都是全球AI领域的一场盛宴&#xff0c;吸引着来自世界各地的顶…
阅读更多...
软件文档-总体测试计划书（Word原件2024）

软件文档-总体测试计划书（Word原件2024）

软件资料清单列表部分文档&#xff1a; 工作安排任务书&#xff0c;可行性分析报告&#xff0c;立项申请审批表&#xff0c;产品需求规格说明书&#xff0c;需求调研计划&#xff0c;用户需求调查单&#xff0c;用户需求说明书&#xff0c;概要设计说明书&#xff0c;技术解决…
阅读更多...
第九期济南科创生态大会即将盛大开启，为中小企业注入新动能

第九期济南科创生态大会即将盛大开启，为中小企业注入新动能

【头部财经】近日&#xff0c;备受瞩目的第九期济南科创生态大会即将拉开帷幕。本次大会将聚焦科技创新领域&#xff0c;为参会者提供一个交流合作、资源对接的优质平台。 作为一个专为中小企业打造的科技创新交流平台&#xff0c;济南科创生态大会一直致力于促进中小企业的发…
阅读更多...
【VTKExamples::Rendering】第七期 TestShadows

【VTKExamples::Rendering】第七期 TestShadows

很高兴在雪易的CSDN遇见你 VTK技术爱好者 QQ：870202403 公众号：VTK忠粉 前言 本文分享VTK样例TestShadows，并解析接口vtkRenderer，希望对各位小伙伴有所帮助！ 感谢各位小伙伴的点赞+关注，小易会继续努力分享，一起进步！ 你的点赞就是我的动力(＾Ｕ＾)ノ~ＹＯ …
阅读更多...
解决Android手机无法通过蓝牙给win10 PC传送文件

解决Android手机无法通过蓝牙给win10 PC传送文件

&#xff08;一&#xff09;先配对设备&#xff0c;正常配对就可以 &#xff08;二&#xff09;打开系统设置&#xff0c;win搜索窗口搜索“设置” &#xff08;三&#xff09;搜索“蓝牙” &#xff08;四&#xff09;打开“蓝牙和其他设备”&#xff0c;点击“更多蓝牙设置”…
阅读更多...
【2024年5月备考新增】】 考前篇（2）《官方平台 - 考生模拟练习平台常用操作（一）》

【2024年5月备考新增】】 考前篇（2）《官方平台 - 考生模拟练习平台常用操作（一）》

软考考生常用操作说明 说明：模拟作答系统是旨在让考生熟悉计算机化考试环境和作答方式，模拟作答不保存考生作答 历史记录。考试题型、题量、分值、界面及文字内容以正式考试答题系统为准。 1 如何标记试题、切换试题 2 简答题如何查看历史记录、切换输入法 3 选做题，已作答…
阅读更多...
C++下使用Matplotlib-cpp的一些配置

C++下使用Matplotlib-cpp的一些配置

Aconda3安装使用 https://mirrors.tuna.tsinghua.edu.cn/help/anaconda/下载 切换清华源 conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/ conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/ con…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023