aws emr启动standalone的flink集群

news/2024/11/13 0:51:15/

关键组件

  • Client,代码由客户端获取并做转换,之后提交给JobManger
  • JobManager,对作业进行中央调度管理,获取到要执行的作业后,会进一步处理转换,然后分发任务给众多的TaskManager。
  • TaskManager,数据的处理操作
image-20240105171119363

在emr上自建standalone集群,文件分发脚本xsync

#!/bin/bash
if [ $# -lt 1 ]
thenecho Not Enough Arguement!exit;
fi
nodelist=$(yarn node -list 2> /dev/null | awk 'NR > 2 {print $1}' | awk -F: '{print $1}')
for host in $nodelist
doecho ==================== $host ====================for file in $@doif [ -e $file ]thenpdir=$(cd -P $(dirname $file); pwd)fname=$(basename $file)ssh $host "mkdir -p $pdir"rsync -av $pdir/$fname $host:$pdirelseecho $file does not exists!fidone
done

先进入节点中创建flink文件(root用户)

#!/bin/bash
nodelist=$(yarn node -list 2> /dev/null | awk 'NR > 2 {print $1}' | awk -F: '{print $1}')
for node in $nodelist; doecho "Executing commands on $node"ssh -o StrictHostKeyChecking=no $node "sudo mkdir -p /usr/lib/flink/sudo chown -R hadoop:hadoop /usr/lib/flink"
done

将/usr/lib/flink/同步到其他节点上

xsync /usr/lib/flink
xsync /etc/flink/conf.dist

重新软连接

#!/bin/bash
nodelist=$(yarn node -list 2> /dev/null | awk 'NR > 2 {print $1}' | awk -F: '{print $1}')
for node in $nodelist; doecho "Executing commands on $node"ssh -o StrictHostKeyChecking=no $node "sudo rm /usr/lib/flink/confsudo ln -s /etc/flink/conf.dist/ /usr/lib/flink/conf"
done

直接启动失败,需要指定workfile

No workers file. Please specify workers in 'conf/workers'.

root用户下写入workers配置

yarn node -list 2> /dev/null | awk 'NR > 2 {print $1}' | awk -F: '{print $1}' > /usr/lib/flink/conf/workersecho $(hostname).cn-north-1.compute.internal:8081 > /usr/lib/flink/conf/masters

jobmanager配置

# JobManager 节点地址.
jobmanager.rpc.address: $(hostname).cn-north-1.compute.internal
jobmanager.bind-host: 0.0.0.0 # default
rest.address: 0.0.0.0 # webui
rest.bind-address: 0.0.0.0

taskmanager配置

echo taskmanager.host: $(hostname).cn-north-1.compute.internal >> /usr/lib/flink/conf/flink-conf.yaml
echo taskmanager.bind-host: 0.0.0.0 >> /usr/lib/flink/conf/flink-conf.yaml

修改权限

sudo chown -R hadoop:hadoop /var/lib/flink

启动集群

  • 不知道为什么worker节点始终无法启动taskmanagerrunner,最终发现没有权限

  • 这个路径是提供webui上传jar文件用的

    mkdir: cannot create directory ‘/var/run/flink’: Permission denied
    /usr/lib/flink/bin/flink-daemon.sh: line 82: /var/run/flink: No such file or directory
    flock: 200: Bad file descriptor
    Starting taskexecutor daemon on host ip-192-168-28-247.
    /usr/lib/flink/bin/flink-daemon.sh: line 145: /var/run/flink/flink-hadoop-taskexecutor.pid: No such file or directory
    
/usr/lib/flink/bin/start-cluster.sh

在master上查看日志

在这里插入图片描述

最终taskmanager成功注册

image-20240105201137482

关闭集群

/usr/lib/flink/bin/stop-cluster.sh

jpsall脚本

#!/bin/bash
masternode=$(hostname).cn-north-1.compute.internal
nodelist=$(yarn node -list 2> /dev/null | awk 'NR > 2 {print $1}' | awk -F: '{print $1}')
for host in $masternode $nodelist
doecho =============== $host ===============ssh $host jps
done

在yarn模式下,提交 JAR 文件后,它就会变成由 Flink JobManager 管理的作业。

  • JobManager 位于托管 Flink 会话 Application Master 进程守护程序的 YARN 节点上

集群生命周期大于job,因此实际上对应session模式

flink run --jobmanager localhost:8081 /usr/lib/flink/examples/streaming/WordCount.jar --input s3://zhaojiew-tmp/shakespeare/ --output s3://zhaojiew-tmp/flinkoutput flink run --jobmanager localhost:8081 -c org.example.wc.WordCountBatch flinkall-1.0.0.jar

部署模式

session模式

启动flink session

  • 5.5.0 版本中添加了 flink-yarn-session 命令作为 yarn-session.sh 脚本的包装程序以简化执行
flink-yarn-session -d

image-20240105110801556

启动session后提交任务

flink run --jobmanager yarn-cluster -yid application_1704427099392_0001  /usr/lib/flink/examples/streaming/WordCount.jar --input s3://zhaojiew-tmp/shakespeare/ --output s3://zhaojiew-tmp/flinkoutput Caused by: software.amazon.awssdk.services.s3.model.S3Exception: null (Service: S3, Status Code: 400, Request ID: 8XKQF6W01QDQCRTD, Extended Request ID: Bsi6AK3alrxV5OYL5aZhu05h/RusTGUBm9P9hRu5dFu0whCv68DKpvFjf8CYL9Wc5zSEoaL759M=)

可以设置region

  • 看起来是6.15版本的emr_flink存在问题,默认region不对
-Dfs.s3a.bucket.endpoint.region=cn-north-1

在resourcemanager的Tracking UI会跳转到flink jobmanager

但是flink的taskmanager日志并没有汇聚到yarn历史服务器中

查看session fluster的ip和端口号

image-20240105222430792

在这里插入图片描述

或者直接在提交jar界面看ip和端口地址,监听的是内网ip

  • 找到后可以在idea中注册

per-job模式

直接运行batch任务

flink run -m yarn-cluster -Dexecution.runtime-mode=BATCH flinktutorial17-1.0.jar

不同类型任务的name

image-20240105155322448

application模式

提交任务

  • 将jar拷贝到/usr/lib/flink/lib

  • 指定作业入口类,脚本会到 lib 目录扫描所有的 jar 包

/usr/lib/flink/bin/standalone-job.sh start --job-classname org.example.wc.WordCountBatchStarting standalonejob daemon on host ip-192-168-30-184.

启动taskmanager

/usr/lib/flink/bin/taskmanager.sh start

在application模式下,yarn中同样没有记录

查看jps进程

image-20240105220805065

在master上查看日志

image-20240105220010908

yarn运行模式

yarn运行模式下同样可以使用三种部署模式

session模式

flink-yarn-session -d命令参数

  • Flink1.11.0 版本不再使用-n 参数和-s 参数分别指定 TaskManager 数量和 slot 数量,YARN 会按照需求动态分配 TaskManager 和 slot
-d:分离模式
-jm(--jobManagerMemory):配置 JobManager 所需内存,默认单位 MB
-nm(--name):配置在 YARN UI 界面上显示的任务名
-qu(--queue):指定 YARN 队列名
-tm(--taskManager):配置每个 TaskManager 所使用内存。

per-job模式

提交任务

/usr/lib/flink/bin/flink run -t yarn-per-job -c org.example.wc.WordCountBatch flinkall-1.0.0.jar-t,--target <arg>     The deployment target for the given application,which is equivalent to the "execution.target" configoption. For the "run" action the currently availabletargets are: "remote", "local", "kubernetes-session","yarn-per-job" (deprecated), "yarn-session". For the"run-application" action the currently availabletargets are: "kubernetes-application","yarn-application".

如下报错的解决

  • flink 的/opt/module/flink-1.17.0/conf/flink-conf.yaml 配置文件中设置classloader.check-leaked-classloader: false

image-20240105223157373

application模式

提交任务

/usr/lib/flink/bin/flink run-application -t yarn-application -c org.example.wc.WordCountBatch flinkall-1.0.0.jar/usr/lib/flink/bin/flink run-application -t yarn-application s3://zhaojiew-bigdata/app/WordCount.jar --input s3://zhaojiew-tmp/shakespeare/ --output s3://zhaojiew-tmp/flinkoutput 

可以看到print输出此时在jobmanager中

image-20240105223507271

但是看起来并不支持将jar存储在s3?已知问题,怀疑和flink本身有关,因为使用s3作为input和output是没有问题的

在这里插入图片描述

可以在提交任务时指定依赖,并非任务jar

bin/flink run-application -t yarnapplication -Dyarn.provided.lib.dirs="hdfs://hadoop102:8020/flinkdist" -c com.atguigu.wc.SocketStreamWordCount
hdfs://hadoop102:8020/flink-jars/FlinkTutorial-1.0-SNAPSHOT.jar

http://www.ppmy.cn/news/1466555.html

相关文章

C++之map

1、标准库的map类型 2、插入数据 #include <map> #include <string> #include <iostream>using namespace std;int main() {map<string, int> mapTest;// 插入到map容器内部的元素是默认按照key从小到大来排序// key类型一定要重载小于号<运算符map…

告别盲目推广!Xinstall二维码携参技术,让App运营更精准高效

在移动互联网时代&#xff0c;App推广和运营已成为每个开发者必须面对的重要任务。然而&#xff0c;如何精准地定位目标用户&#xff0c;提高转化率和用户留存率&#xff0c;成为了摆在每个开发者面前的难题。今天&#xff0c;我们就来谈谈如何通过Xinstall二维码携参技术&…

Linux 内核之 mmap 内存映射触发的缺页异常 Page Fault

文章目录 前言一、简介1. MMU 内存管理2. 缺页中断3. 页表4. 小节 二、mmap 提前分配物理内存1. mm_populate 函数2. __mm_populate 函数3. populate_vma_page_range 函数4. __get_user_pages 函数5. find_extend_vma 函数6. find_vma 函数7. follow_page_mask 函数8. follow_p…

钉钉企业内部H5微应用或小程序之钉消息推送

钉钉简单的推送钉消息 一、钉钉准备工作 首先进入钉钉开放平台 你得有企业内部微应用或者小程序 没有创建的话去看我另一篇文章有说明 钉钉开放平台创建企业内部H5微应用或者小程序-CSDN博客 看不懂话也可以参考官方文档&#xff1a;创建应用 - 钉钉开放平台 二、开发的准备…

低代码开发:助企构建数字化应用平台

随着信息技术的快速发展&#xff0c;数字化应用平台已经成为企业提升竞争力、实现业务创新的关键。然而&#xff0c;传统的应用开发方式往往面临着开发周期长、成本高昂、技术门槛高等问题&#xff0c;这使得许多企业望而却步。而低代码开发技术的出现&#xff0c;为企业构建数…

算法思想总结:哈希表

一、哈希表剖析 1、哈希表底层&#xff1a;通过对C的学习&#xff0c;我们知道STL中哈希表底层是用的链地址法封装的开散列。 2、哈希表作用&#xff1a;存储数据的容器&#xff0c;插入、删除、搜索的时间复杂度都是O&#xff08;1&#xff09;&#xff0c;无序。 3、什么时…

GPT-4o:人工智能新纪元的开端

引言 近年来&#xff0c;人工智能领域的发展日新月异&#xff0c;特别是在自然语言处理&#xff08;NLP&#xff09;领域&#xff0c;各种生成预训练模型不断推陈出新。自OpenAI发布GPT-3以来&#xff0c;生成预训练模型在文本生成、语言理解等任务中展现了强大的能力。近期&a…

《探索Stable Diffusion:AI绘画的创意之路与实战秘籍》

《Stable Diffusion AI 绘画从提示词到模型出图》介绍了 Stable Diffusion AI 绘画工具及其使用技巧。书中内容分为两部分&#xff1a;“基础操作篇”&#xff0c;讲解了 SD 文生图、图生图、提示词、模型、ControlNet 插件等核心技术的应用&#xff0c;帮助读者快速从新手成长…