【极数系列】Flink配置参数如何获取？（06）

文章目录

  • gitee码云地址
  • 简介概述
  • 01 配置值来自.properties文件
    • 1.通过路径读取
    • 2.通过文件流读取
    • 3.通过IO流读取
  • 02 配置值来自命令行
  • 03 配置来自系统属性
  • 04 注册以及使用全局变量
  • 05 Flink获取参数值Demo
    • 1.项目结构
    • 2.pom.xml文件如下
    • 3.配置文件
    • 4.项目主类
    • 5.运行查看相关日志

gitee码云地址

直接下载解压可用 https://gitee.com/shawsongyue/aurora.git
模块：aurora_flink
主类：GetParamsStreamingJob

简介概述

​ 1.几乎所有的批和流的 Flink 应用程序，都依赖于外部配置参数。这些配置参数可以用于指定输入和输出源（如路径或地址）、系统参数（并行度，运行时配置）和特定的应用程序参数（通常使用在用户自定义函数）。

​ 2.为解决以上问题，Flink 提供一个名为 Parametertool 的简单公共类，其中包含了一些基本的工具。请注意，这里说的 Parametertool 并不是必须使用的。Commons CLI 和 argparse4j 等其他框架也可以非常好地兼容 Flink。

​ 3.**ParameterTool**定义了一组静态方法，用于读取配置信息。该工具类内部使用了 Map` 类型，这样使得它可以很容易地与你的配置集成在一起。

01 配置值来自.properties文件

1.通过路径读取

//定义文件路径
String propertiesFilePath = "E:\\project\\aurora_dev\\aurora_flink\\src\\main\\resources\\application.properties";//方式一:直接使用内置工具类
ParameterTool parameter_01 = ParameterTool.fromPropertiesFile(propertiesFilePath);
String jobName_01 = parameter_01.get("jobName");
logger.info("方式一:读取配置文件中指定的key值={}",jobName_01);

2.通过文件流读取

//定义文件路径
String propertiesFilePath = "E:\\project\\aurora_dev\\aurora_flink\\src\\main\\resources\\application.properties";//方式二：使用文件
File propertiesFile = new File(propertiesFilePath);
ParameterTool parameter_02 = ParameterTool.fromPropertiesFile(propertiesFile);
String jobName_02 = parameter_02.get("jobName");
logger.info("方式二:读取配置文件中指定的key值={}",jobName_02);

3.通过IO流读取

//定义文件路径
String propertiesFilePath = "E:\\project\\aurora_dev\\aurora_flink\\src\\main\\resources\\application.properties";//方式三：使用IO流
InputStream propertiesFileInputStream = new FileInputStream(new File(propertiesFilePath));
ParameterTool parameter_03 = ParameterTool.fromPropertiesFile(propertiesFileInputStream);
String jobName_03 = parameter_03.get("jobName");
logger.info("方式三:读取配置文件中指定的key值={}",jobName_03);

02 配置值来自命令行

tips：在idea的命令行传参，格式：–jobName program_job_aurora

在这里插入图片描述

ParameterTool parameter_04 = ParameterTool.fromArgs(args);
String jobName_04 = parameter_04.get("jobName");
logger.info("方式四:命令行传参key值={}",jobName_04);

03 配置来自系统属性

tips：在idea的的jvm系统参数设置，格式：-Dinput=hdfs:///mydata

在这里插入图片描述

//方式五：获取jvm参数值
ParameterTool parameter_05 = ParameterTool.fromSystemProperties();
String jobName_05 = parameter_05.get("input");
logger.info("方式五:获取jvm参数key值={}",jobName_05);

04 注册以及使用全局变量

注意：Flink全局变量仅支持在富函数中使用，即Rich开头的类使用

//定义文件路径
String propertiesFilePath = "E:\\project\\aurora_dev\\aurora_flink\\src\\main\\resources\\application.properties";//直接使用内置工具类获取参数
ParameterTool parameter_01 = ParameterTool.fromPropertiesFile(propertiesFilePath);//方式六：注册全局参数final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.getConfig().setGlobalJobParameters(parameter_01);//在任意富函数中均可以获取,注意！注意！注意！只有富文本函数才可以使用//1.创建富函数RichFlatMapFunction<String, String> richFlatMap = new RichFlatMapFunction<>() {@Overridepublic void flatMap(String s, Collector<String> collector) throws Exception {//获取运行环境ParameterTool parameters = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();//获取对应的值String jobName = parameters.getRequired("jobName");logger.info("方式六:获取全局注册参数key值={}",jobName_05);}};//2.创建数据集ArrayList<String> list = new ArrayList<>();list.add("001");list.add("002");list.add("003");//3.把有限数据集转换为数据源DataStreamSource<String> dataStreamSource = env.fromCollection(list).setParallelism(1);//4.执行富文本处理dataStreamSource.flatMap(richFlatMap);//5.启动程序env.execute();

05 Flink获取参数值Demo

1.项目结构

在这里插入图片描述

2.pom.xml文件如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.xsy</groupId><artifactId>aurora_flink</artifactId><version>1.0-SNAPSHOT</version><!--属性设置--><properties><!--java_JDK版本--><java.version>11</java.version><!--maven打包插件--><maven.plugin.version>3.8.1</maven.plugin.version><!--编译编码UTF-8--><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><!--输出报告编码UTF-8--><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><!--json数据格式处理工具--><fastjson.version>1.2.75</fastjson.version><!--log4j版本--><log4j.version>2.17.1</log4j.version><!--flink版本--><flink.version>1.18.0</flink.version><!--scala版本--><scala.binary.version>2.11</scala.binary.version><!--log4j依赖--><log4j.version>2.17.1</log4j.version></properties><!--通用依赖--><dependencies><!-- json --><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.12</artifactId><version>${flink.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version></dependency><!--================================集成外部依赖==========================================--><!--集成日志框架 start--><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-slf4j-impl</artifactId><version>${log4j.version}</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-api</artifactId><version>${log4j.version}</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><version>${log4j.version}</version></dependency><!--集成日志框架 end--></dependencies><!--编译打包--><build><finalName>${project.name}</finalName><!--资源文件打包--><resources><resource><directory>src/main/resources</directory></resource><resource><directory>src/main/java</directory><includes><include>**/*.xml</include></includes></resource></resources><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>3.1.1</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><artifactSet><excludes><exclude>org.apache.flink:force-shading</exclude><exclude>org.google.code.flindbugs:jar305</exclude><exclude>org.slf4j:*</exclude><excluder>org.apache.logging.log4j:*</excluder></excludes></artifactSet><filters><filter><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>org.xsy.sevenhee.flink.TestStreamJob</mainClass></transformer></transformers></configuration></execution></executions></plugin></plugins><!--插件统一管理--><pluginManagement><plugins><!--maven打包插件--><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><version>${spring.boot.version}</version><configuration><fork>true</fork><finalName>${project.build.finalName}</finalName></configuration><executions><execution><goals><goal>repackage</goal></goals></execution></executions></plugin><!--编译打包插件--><plugin><artifactId>maven-compiler-plugin</artifactId><version>${maven.plugin.version}</version><configuration><source>${java.version}</source><target>${java.version}</target><encoding>UTF-8</encoding><compilerArgs><arg>-parameters</arg></compilerArgs></configuration></plugin></plugins></pluginManagement></build><!--配置Maven项目中需要使用的远程仓库--><repositories><repository><id>aliyun-repos</id><url>https://maven.aliyun.com/nexus/content/groups/public/</url><snapshots><enabled>false</enabled></snapshots></repository></repositories><!--用来配置maven插件的远程仓库--><pluginRepositories><pluginRepository><id>aliyun-plugin</id><url>https://maven.aliyun.com/nexus/content/groups/public/</url><snapshots><enabled>false</enabled></snapshots></pluginRepository></pluginRepositories></project>

3.配置文件

（1）application.properties

jobName=job_aurora
jobMemory=1024
taskName=task_aurora

（2）log4j2.properties

rootLogger.level=INFO
rootLogger.appenderRef.console.ref=ConsoleAppender
appender.console.name=ConsoleAppender
appender.console.type=CONSOLE
appender.console.layout.type=PatternLayout
appender.console.layout.pattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
log.file=D:\\tmprootLogger.level=INFO
rootLogger.appenderRef.console.ref=ConsoleAppender
appender.console.name=ConsoleAppender
appender.console.type=CONSOLE
appender.console.layout.type=PatternLayout
appender.console.layout.pattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
log.file=D:\\tmp

4.项目主类

package com.aurora;import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;/*** @description flink获取外部参数作业** @author 浅夏的猫* @datetime 15:54 2024/1/28
*/
public class GetParamsStreamingJob {private static final Logger logger = LoggerFactory.getLogger(GetParamsStreamingJob.class);public static void main(String[] args) throws Exception {//定义文件路径String propertiesFilePath = "E:\\project\\aurora_dev\\aurora_flink\\src\\main\\resources\\application.properties";//方式一:直接使用内置工具类ParameterTool parameter_01 = ParameterTool.fromPropertiesFile(propertiesFilePath);String jobName_01 = parameter_01.get("jobName");logger.info("方式一:读取配置文件中指定的key值={}",jobName_01);//方式二：使用文件File propertiesFile = new File(propertiesFilePath);ParameterTool parameter_02 = ParameterTool.fromPropertiesFile(propertiesFile);String jobName_02 = parameter_02.get("jobName");logger.info("方式二:读取配置文件中指定的key值={}",jobName_02);//方式三：使用IO流InputStream propertiesFileInputStream = new FileInputStream(new File(propertiesFilePath));ParameterTool parameter_03 = ParameterTool.fromPropertiesFile(propertiesFileInputStream);String jobName_03 = parameter_03.get("jobName");logger.info("方式三:读取配置文件中指定的key值={}",jobName_03);//方式四：命令行传参格式：--jobName program_job_auroraParameterTool parameter_04 = ParameterTool.fromArgs(args);String jobName_04 = parameter_04.get("jobName");logger.info("方式四:命令行传参key值={}",jobName_04);//方式五：获取jvm参数值ParameterTool parameter_05 = ParameterTool.fromSystemProperties();String jobName_05 = parameter_05.get("input");logger.info("方式五:获取jvm参数key值={}",jobName_05);//方式六：注册全局参数final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.getConfig().setGlobalJobParameters(parameter_01);//在任意富函数中均可以获取,注意！注意！注意！只有富文本函数才可以使用//1.创建富函数RichFlatMapFunction<String, String> richFlatMap = new RichFlatMapFunction<>() {@Overridepublic void flatMap(String s, Collector<String> collector) throws Exception {//获取运行环境ParameterTool parameters = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();//获取对应的值String jobName = parameters.getRequired("jobName");logger.info("方式六:获取全局注册参数key值={}",jobName_05);}};//2.创建数据集ArrayList<String> list = new ArrayList<>();list.add("001");list.add("002");list.add("003");//3.把有限数据集转换为数据源DataStreamSource<String> dataStreamSource = env.fromCollection(list).setParallelism(1);//4.执行富文本处理dataStreamSource.flatMap(richFlatMap);//5.启动程序env.execute();}}

5.运行查看相关日志

在这里插入图片描述

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

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

相关文章

Linux 网络流量相关工具

Linux 网络流量相关工具

本文聚焦于网络流量的查看、端口占用查看。至于网络设备的管理和配置&#xff0c;因为太过复杂且不同发行版有较大差异&#xff0c;这里就不赘述&#xff0c;后面看情况再写。 需要注意的是&#xff0c;这里列出的每一个工具都有丰富的功能&#xff0c;流量/端口信息查看只是其…
阅读更多...
探索Pyecharts之美-绘制多彩旭日图的艺术与技巧【第37篇—python：旭日图】

探索Pyecharts之美-绘制多彩旭日图的艺术与技巧【第37篇—python：旭日图】

文章目录 引言准备工作绘制基本旭日图调整颜色和样式添加交互功能定制标签和标签格式嵌套层级数据高级样式与自定义进阶主题&#xff1a;动态旭日图数据源扩展&#xff1a;外部JSON文件总结 引言 数据可视化在现代编程中扮演着重要的角色&#xff0c;而Pyecharts是Python中一个…
阅读更多...
数据结构——链式二叉树（3）

数据结构——链式二叉树（3）

本篇文章我们依然讲解链式二叉树的OJ题&#xff1b; 一、二叉树的层序遍历 层序遍历即从根节点开始一层一层的遍历。我们可以运用队列的先进先出特性实现&#xff01; //层序遍历 void a(BTNode* root) {Que qhead;Queueinit(&qhead);//先入队根节点if(root)QueuePush(&…
阅读更多...
三维重建（7）--运动恢复结构SfM系统解析

三维重建（7）--运动恢复结构SfM系统解析

目录 一、SfM系统&#xff08;两视图&#xff09; 1、特征提取 2、特征匹配 3、RANSAC求解基础矩阵F 4、完整的欧式结构恢复算法流程 二、基于增量法的SfM系统&#xff08;以OpenMVG为例&#xff09; 1、预处理 2、图像特征点提取与匹配 3、两视图重构点云 4、增加…
阅读更多...
LPC系列一个定时器不同频率

LPC系列一个定时器不同频率

1.背景 最近研究的LPC804里只有一个ctimer&#xff0c;很多时候用的捉襟见肘的&#xff0c;官方给了一份双匹配的参考例程&#xff0c;不过实际用处不大。不过我花了一晚上的时间&#xff0c;终于研究出来将一个定时器拆成四个定时器用的办法了。这个方法适用于用回调函数的LP…
阅读更多...
RabbitMQ（一）

RabbitMQ（一）

1、相关概念 1.1、消息队列&#xff08;MQ&#xff09; MQ(message queue)&#xff0c;从字面意思上看&#xff0c;本质是个队列&#xff0c;FIFO 先入先出&#xff0c;只不过队列中存放的内容是message 而已&#xff0c;还是一种跨进程的通信机制&#xff0c;用于上下游传递消…
阅读更多...
移动Web——平面转换-多重转换

移动Web——平面转换-多重转换

1、平面转换-多重转换 多重转换技巧&#xff1a;先平移再旋转 <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8" /><meta http-equiv"X-UA-Compatible" content"IEedge" /><meta name&qu…
阅读更多...
数据结构——链式二叉树

数据结构——链式二叉树

目录 &#x1f341;一、二叉树的遍历 &#x1f315;&#xff08;一&#xff09;、前序遍历(Preorder Traversal 亦称先序遍历) &#x1f315;&#xff08;二&#xff09;、中序遍历(Inorder Traversal) &#x1f315;&#xff08;三&#xff09;、后序遍历(Postorder Traver…
阅读更多...
scrapy的入门使用

scrapy的入门使用

1 安装scrapy 命令: sudo apt-get install scrapy或者&#xff1a; pip/pip3 install scrapy2 scrapy项目开发流程 创建项目: scrapy startproject mySpider生成一个爬虫: scrapy genspider itcast itcast.cn提取数据:     根据网站结构在spider中实现数据采集相关内…
阅读更多...
centos系统安装Ward服务器监控工具

centos系统安装Ward服务器监控工具

简介 Ward是一个简约美观多系统支持的服务器监控面板 安装 1.首先安装jdk yum install java-1.8.0-openjdk-devel.x86_64 2.下载jar wget 3.启动 java -jar ward-1.8.8.jar 体验 浏览器输入 http://192.168.168.110:4000/ 设置服务名设置为:myserver 端口号:5000 点击…
阅读更多...
WSL2 Debian系统添加支持SocketCAN

WSL2 Debian系统添加支持SocketCAN

本人最近在使用WSL2&#xff0c;Linux系统选择的是Debian&#xff0c;用起来很不错&#xff0c;感觉可以代替VMware Player虚拟机。 但是WSL2 Debian默认不支持SocketCAN&#xff0c;这就有点坑了&#xff0c;由于本人经常要使用SocketCAN功能&#xff0c;所以决定让Debian支持…
阅读更多...
开始学习第二十五天（番外）

开始学习第二十五天（番外）

今天分享一下写的小游戏啦 头文件game.h #include<stdio.h> #include<time.h> #include<stdlib.h> #define H 3 #define L 3 void InitBoard(char Board[H][L], int h, int l); void DisplayBoard(char Board[H][L], int h, int l); void playermove(cha…
阅读更多...
【LeetCode: Z 字形变换 + 模拟】

【LeetCode: Z 字形变换 + 模拟】

&#x1f680; 算法题 &#x1f680; &#x1f332; 算法刷题专栏 | 面试必备算法 | 面试高频算法 &#x1f340; &#x1f332; 越难的东西,越要努力坚持&#xff0c;因为它具有很高的价值&#xff0c;算法就是这样✨ &#x1f332; 作者简介&#xff1a;硕风和炜&#xff0c;…
阅读更多...
5G_RACH（一）

5G_RACH（一）

什么是RACH RACH 代表 Random Access Channel。这是开机时UE发给eNB的第一条消息。 为什么选择RACH &#xff1f;&#xff08;RACH 的功能是什么&#xff1f; 当你第一次听到RACH或RACH Process这个词时&#xff0c;你脑海中浮现的第一个问题是“为什么是RACH&#xff1f;”…
阅读更多...
Windows XP x86 sp3 安装 Google Chrome 49.0.2623.112 (正式版本) （32 位）

Windows XP x86 sp3 安装 Google Chrome 49.0.2623.112 (正式版本) （32 位）

1 下载地址&#xff1b; https://dl.google.com/release2/h8vnfiy7pvn3lxy9ehfsaxlrnnukgff8jnodrp0y21vrlem4x71lor5zzkliyh8fv3sryayu5uk5zi20ep7dwfnwr143dzxqijv/49.0.2623.112_chrome_installer.exe 2 直接 双击 49.0.2623.112_chrome_installer.exe 安装&#xff1b; 3 …
阅读更多...
Redis6基础知识梳理~

Redis6基础知识梳理~

初识NOSQL&#xff1a; NOSQL是为了解决性能问题而产生的技术&#xff0c;在最初&#xff0c;我们都是使用单体服务器架构&#xff0c;如下所示&#xff1a; 随着用户访问量大幅度提升&#xff0c;同时产生了大量的用户数据&#xff0c;单体服务器架构面对着巨大的压力 NOSQL解…
阅读更多...
SpringBoot之JWT登录

SpringBoot之JWT登录

JWT JSON Web Token&#xff08;JSON Web令牌&#xff09; 是一个开放标准(rfc7519)&#xff0c;它定义了一种紧凑的、自包含的方式&#xff0c;用于在各方之间以JSON对象安全地传输信息。此信息可以验证和信任&#xff0c;因为它是数字签名的。jwt可以使用秘密〈使用HNAC算法…
阅读更多...
10. UE5 RPG使用GameEffect创建血瓶修改角色属性

10. UE5 RPG使用GameEffect创建血瓶修改角色属性

前面我们通过代码实现了UI显示角色的血量和蓝量&#xff0c;并实现了初始化和在数值变动时实时更新。为了测试方便&#xff0c;没有使用GameEffect去修改角色的属性&#xff0c;而是通过代码直接修改的数值。 对于GameEffect的基础&#xff0c;这里不再讲解&#xff0c;如果需要…
阅读更多...
《动手学深度学习(PyTorch版)》笔记4.4

《动手学深度学习(PyTorch版)》笔记4.4

注&#xff1a;书中对代码的讲解并不详细&#xff0c;本文对很多细节做了详细注释。另外&#xff0c;书上的源代码是在Jupyter Notebook上运行的&#xff0c;较为分散&#xff0c;本文将代码集中起来&#xff0c;并加以完善&#xff0c;全部用vscode在python 3.9.18下测试通过。…
阅读更多...
鸿蒙（HarmonyOS）项目方舟框架（ArkUI）之Swiper容器组件

鸿蒙（HarmonyOS）项目方舟框架（ArkUI）之Swiper容器组件

鸿蒙&#xff08;HarmonyOS&#xff09;项目方舟框架&#xff08;ArkUI&#xff09;之Swiper容器组件 一、操作环境 操作系统: Windows 10 专业版、IDE:DevEco Studio 3.1、SDK:HarmonyOS 3.1 二、Swiper容器组件 滑块视图容器&#xff0c;提供子组件滑动轮播显示的能力。…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023