Hadoop学习笔记（HDP）-Part.18 安装Flink

news/2024/7/7 22:43:33

目录
Part.01 关于HDP
Part.02 核心组件原理
Part.03 资源规划
Part.04 基础环境配置
Part.05 Yum源配置
Part.06 安装OracleJDK
Part.07 安装MySQL
Part.08 部署Ambari集群
Part.09 安装OpenLDAP
Part.10 创建集群
Part.11 安装Kerberos
Part.12 安装HDFS
Part.13 安装Ranger
Part.14 安装YARN+MR
Part.15 安装HIVE
Part.16 安装HBase
Part.17 安装Spark2
Part.18 安装Flink
Part.19 安装Kafka
Part.20 安装Flume

十八、安装Flink

1.配置Ambari的flink资源

(1)创建flink源

下载链接为
https://repo.huaweicloud.com/apache/flink/flink-1.9.3/flink-1.9.3-bin-scala_2.12.tgz
https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.6.5-7.0/flink-shaded-hadoop-2-uber-2.6.5-7.0.jar
http://www.java2s.com/Code/JarDownload/javax.ws/javax.ws.rs-api-2.0.jar.zip
上传到hdp01上，并复制到/var/www/html下

mkdir /var/www/html/flink
cp /opt/flink-1.9.3-bin-scala_2.12.tgz /var/www/html/flink/
cp /opt/flink-shaded-hadoop-2-uber-2.6.5-7.0.jar /var/www/html/flink/
cp /opt/javax.ws.rs-api-2.0.jar /var/www/html/flink/

(2)下载ambari-flink-service服务

在外网服务器上

git clone https://ghproxy.com/https://gitee.com/liujingwen-git/ambari-flink-service-master.git /root/FLINK

注：github.com无法直接下载，可以使用gproxy.com进行代理加速，拼接形成加速URL；或者将github.com在本地强制解析为140.82.114.4
将FLINK文件夹上传到hdp01的/var/lib/ambari-server/resources/stacks/HDP/3.1/services/上，对应的版本可在集群的服务器中查看

hdp-select status hadoop-client | sed 's/hadoop-client - \([0-9]\.[0-9]\).*/\1/'

(3)文件说明

从github上下载的ambari-flink-service文件较多，有些可以删除以及修改
在这里插入图片描述

(4)修改metainfo.xml文件

文件位置：
/var/lib/ambari-server/resources/stacks/HDP/3.1/services/FLINK/metainfo.xml
修改版本

            <name>FLINK</name>
            <displayName>Flink</displayName>
            <comment>Apache Flink is a streaming dataflow engine that provides data distribution, communication, and fault tolerance for distributed computations over data streams.</comment>
            <version>1.9.3</version>

删除FLINK_MASTER，仅保留安装FLINK_CLIENT，在Flink on YARN模式下，master与ResourceManager合设，无需单独安装

            <components>
                <component>
                  <name>FLINK_CLIENT</name>
                  <displayName>FlinkCLIENT</displayName>
                  <category>CLIENT</category>
                  <cardinality>1+</cardinality>
                  <commandScript>
                    <script>scripts/flink_client.py</script>
                    <scriptType>PYTHON</scriptType>
                    <timeout>10000</timeout>
                  </commandScript>
                </component>
            </components>

(5)修改flink-ambari-config.xml文件

文件位置：
/var/lib/ambari-server/resources/stacks/HDP/3.1/services/FLINK/configuration/flink-ambari-config.xml
修改安装路径

  <property>
    <name>flink_install_dir</name>
    <value>/usr/hdp/3.1.5.0-152/flink</value>
    <description>Location to install Flink</description>
  </property>

修改安装包下载地址

  <property>
    <name>flink_download_url</name>
    <value>http://hdp01.hdp.com/flink/flink-1.9.3-bin-scala_2.12.tgz</value>
    <description>Snapshot download location. Downloaded when setup_prebuilt is true</description>
  </property>

  <property>
    <name>flink_hadoop_shaded_jar</name>
    <value>http://hdp01.hdp.com/flink/flink-shaded-hadoop-2-uber-2.6.5-7.0.jar</value>
    <description>Flink shaded hadoop jar download location. Downloaded when setup_prebuilt is true</description>
  </property>
</configuration>

(6)修改flink-env.xml文件

文件位置：
/var/lib/ambari-server/resources/stacks/HDP/3.1/services/FLINK/configuration/flink-env.xml
修改JAVA环境变量

env.java.home: /usr/local/jdk1.8.0_351/jre/

(7)修改flink_client.py文件

文件位置：
/var/lib/ambari-server/resources/stacks/HDP/3.1/services/FLINK/package/scripts/flink_client.py
该文件的作用调用相关的环境变量，来实现整个安装过程
内容如下：

#!/usr/bin/env ptyhon

from resource_management import *

class FlinkClient(Script):
  def install(self, env):
    print 'Install the Flink Client'
    import params
    env.set_params(params)
    self.configure(env)
    # create flink log dir
    Directory(params.flink_log_dir,
              owner=params.flink_user,
              group=params.flink_group,
              create_parents=True,
              mode=0775
             )
    Execute(format("rm -rf {flink_install_dir}/log/flink"))
    Execute(format("ln -s {flink_log_dir} {flink_install_dir}/log/flink"))

  def configure(self, env):
    import params
    env.set_params(params)
    # write out flink-conf.yaml
    properties_content = InlineTemplate(params.flink_yaml_content)
    File(format(params.flink_install_dir + "/conf/flink-conf.yaml"), content=properties_content)

  def status(self, env):
    raise ClientComponentHasNoStatus()

if __name__ == "__main__":
  FlinkClient().execute()

(8)修改parameter.py文件

文件位置：
/var/lib/ambari-server/resources/stacks/HDP/3.1/services/FLINK/package/scripts/params.py
该文件定义了部分环境变量
内容如下：

#!/usr/bin/env python
from resource_management import *
from resource_management.libraries.script.script import Script
import sys, os, glob
from resource_management.libraries.functions.version import format_stack_version
from resource_management.libraries.functions.default import default

# server configurations
config = Script.get_config()

# params from flink-ambari-config
flink_install_dir = config['configurations']['flink-ambari-config']['flink_install_dir']
flink_numcontainers = config['configurations']['flink-ambari-config']['flink_numcontainers']
flink_numberoftaskslots= config['configurations']['flink-ambari-config']['flink_numberoftaskslots']
flink_jobmanager_memory = config['configurations']['flink-ambari-config']['flink_jobmanager_memory']
flink_container_memory = config['configurations']['flink-ambari-config']['flink_container_memory']
setup_prebuilt = config['configurations']['flink-ambari-config']['setup_prebuilt']
flink_appname = config['configurations']['flink-ambari-config']['flink_appname']
flink_queue = config['configurations']['flink-ambari-config']['flink_queue']
flink_streaming = config['configurations']['flink-ambari-config']['flink_streaming']

hadoop_conf_dir = config['configurations']['flink-ambari-config']['hadoop_conf_dir']
flink_download_url = config['configurations']['flink-ambari-config']['flink_download_url']
flink_hadoop_shaded_jar_url = config['configurations']['flink-ambari-config']['flink_hadoop_shaded_jar']
javax_ws_rs_api_jar = config['configurations']['flink-ambari-config']['javax_ws_rs_api_jar']

conf_dir=''
bin_dir=''

# params from flink-conf.yaml
flink_yaml_content = config['configurations']['flink-env']['content']
flink_user = config['configurations']['flink-env']['flink_user']
flink_group = config['configurations']['flink-env']['flink_group']
flink_log_dir = config['configurations']['flink-env']['flink_log_dir']
flink_log_file = os.path.join(flink_log_dir,'flink-setup.log')

temp_file='/tmp/flink.tgz'

(9)添加用户及组

添加用户和组

groupadd flink
useradd -d /home/flink -g flink flink

(10)重启ambari服务

重启服务

ambari-server restart

在ambari中的Stack and Versions中可以看到flink的信息
在这里插入图片描述

2.安装

添加flink服务
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在Custom flink-env中新增
Key：yarn.client.failover-proxy-provider
Value：org.apache.hadoop.yarn.client.ConfiguredRMFailoverProxyProvider
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

3.实验：wordcount

实验：wordcount
从尚硅谷下载实验用的程序代码，在idea中对以socket形式接收数据流的代码进行修改，从192.168.111.1的nc处接收数据流，然后对词频统计后输出到本地文件中
在这里插入图片描述

public static void main(String[] args) throws Exception {
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    DataStream<String> stream = env.socketTextStream("192.168.111.1", 1234);
    DataStream<Tuple2<String, Integer>> resultStream = stream.flatMap(new Tokenizer())
            .keyBy(0)
            .sum(1);
    resultStream.print();
    resultStream.writeAsText("/tmp/wc_result.txt");

    env.execute("Flink Streaming Java API Skeleton");
}

在idea中的Build->Build Artifacts中选择build生成jar包，然后上传到hdp05上，使用一个租户进行kerberos认证后，提交flink任务。

cd /usr/hdp/3.1.5.0-152/flink/bin/
./flink run -m yarn-cluster /root/flink-tutorial-master.jar -c WordCountFromSocket

在这里插入图片描述
在192.168.111.1上启动nc

nc -l 1234

在这里插入图片描述
输入数据流后，中断nc进程，然后在hdp05上查看结果文件/tmp/wc_result.txt
在这里插入图片描述
在yarn上可以查看到对应的应用信息
在这里插入图片描述

4.常见报错

(1)安装时报错Unable to run the custom hook script

报错信息：Error: Error: Unable to run the custom hook script [‘/usr/bin/python’, ‘/var/lib/ambari-agent/cache/stack-hooks/before-ANY/scripts/hook.py’, ‘ANY’, ‘/var/lib/ambari-agent/data/command-1360.json’, ‘/var/lib/ambari-agent/cache/stack-hooks/before-ANY’, ‘/var/lib/ambari-agent/data/structured-out-1360.json’, ‘INFO’, ‘/var/lib/ambari-agent/tmp’, ‘PROTOCOL_TLSv1_2’, ‘’]
2023-03-26 22:23:31,445 - The repository with version 3.1.5.0-152 for this command has been marked as resolved. It will be used to report the version of the component which was installed
在这里插入图片描述
通过ambari添加自定义服务时，总是不能自动增加service账号

python configs.py -u admin -p lnyd@LNsy115 -n HDP315 -l hdp01 -t 8080 -a get -c cluster-env | grep -i ignore_groupsusers_create
python configs.py -u admin -p lnyd@LNsy115 -n HDP315 -l hdp01 -t 8080 -a set -c cluster-env -k ignore_groupsusers_create -v true

在这里插入图片描述

(2)安装后启动报错parent directory /usr/local/flink/conf doesn’t exist

安装后在启动时报错：resource_management.core.exceptions.Fail: Applying File[‘/usr/local/flink/conf/flink-conf.yaml’] failed, parent directory /usr/local/flink/conf doesn’t exist
在这里插入图片描述
不知什么原因没解压过去，需要手动解压到该目录

tar -zxvf /opt/flink-1.9.3-bin-scala_2.12.tgz -C /root/
mv /root/flink-1.9.3/* /usr/local/flink/

(3)提交任务报错java.lang.NoClassDefFoundError

报错信息：
java.lang.NoClassDefFoundError: com/sun/jersey/core/util/FeaturesAndProperties
在这里插入图片描述
maven会自动下载相关的依赖jar包，因此需要将project下的jersey依赖jar包拷贝至flink的lib目录下
在这里插入图片描述


http://lihuaxi.xjx100.cn/news/1869658.html

相关文章

【3】密评-物理和环境安全测评

【3】密评-物理和环境安全测评

0x01 依据 GB/T 39786 -2021《信息安全技术 信息系统密码应用基本要求》针对等保三级系统要求&#xff1a; 物理和环境层面&#xff1a; a&#xff09;宜采用密码技术进行物理访问身份鉴别,保证重要区域进入人员身份的真实性&#xff1b; b&#xff09;宜采用密码技术保证电子门…
阅读更多...
智能优化算法应用：基于学生心理学算法无线传感器网络(WSN)覆盖优化 - 附代码

智能优化算法应用：基于学生心理学算法无线传感器网络(WSN)覆盖优化 - 附代码

智能优化算法应用&#xff1a;基于学生心理学算法无线传感器网络(WSN)覆盖优化 - 附代码 文章目录 智能优化算法应用&#xff1a;基于学生心理学算法无线传感器网络(WSN)覆盖优化 - 附代码1.无线传感网络节点模型2.覆盖数学模型及分析3.学生心理学算法4.实验参数设定5.算法结果…
阅读更多...
WebDriver运行原理的深入剖析

WebDriver运行原理的深入剖析

在现代软件开发中&#xff0c;自动化测试已经成为了不可或缺的一部分。它可以帮助开发者快速、准确地完成软件的功能测试&#xff0c;提高开发效率。而WebDriver就是实现这一目标的重要工具之一。那么&#xff0c;WebDriver是如何工作的呢&#xff1f;本文将通过生活案例&#…
阅读更多...
软著项目推荐 深度学习的智能中文对话问答机器人

软著项目推荐 深度学习的智能中文对话问答机器人

文章目录 0 简介1 项目架构2 项目的主要过程2.1 数据清洗、预处理2.2 分桶2.3 训练 3 项目的整体结构4 重要的API4.1 LSTM cells部分&#xff1a;4.2 损失函数&#xff1a;4.3 搭建seq2seq框架&#xff1a;4.4 测试部分&#xff1a;4.5 评价NLP测试效果&#xff1a;4.6 梯度截断…
阅读更多...
vue3 input 上传 vuedraggable 实现拖拽排序

vue3 input 上传 vuedraggable 实现拖拽排序

效果如下 input 实现上传功能 通过隐藏 input 元素&#xff0c;点击上传触发 input 点击事件&#xff0c;监听 input change 事件 accept 上传文件的类型 multiple 是否允许上传多个 <template><div class"cursor-pointer" click"submitUpload"&…
阅读更多...
Python 中错误 ModuleNotFoundError: No Module Named Configparser

Python 中错误 ModuleNotFoundError: No Module Named Configparser

ModuleNotFoundError 是使用一些需要导入当前程序的内置功能、类、库和包时最常见的错误之一。 大多数情况下&#xff0c;您需要使用 import 关键字将这些内置功能导入到您当前的程序中&#xff1b; 但是&#xff0c;对于高级包和库&#xff0c;您将需要通过命令行界面 (CLI) …
阅读更多...
Hadoop学习笔记（HDP）-Part.10 创建集群

Hadoop学习笔记（HDP）-Part.10 创建集群

目录 Part.01 关于HDP Part.02 核心组件原理 Part.03 资源规划 Part.04 基础环境配置 Part.05 Yum源配置 Part.06 安装OracleJDK Part.07 安装MySQL Part.08 部署Ambari集群 Part.09 安装OpenLDAP Part.10 创建集群 Part.11 安装Kerberos Part.12 安装HDFS Part.13 安装Ranger …
阅读更多...
Python---对象的属性添加与获取

Python---对象的属性添加与获取

1、什么是属性 在Python中&#xff0c;任何一个对象都应该由两部分组成&#xff1a;属性 方法 属性即是特征&#xff0c;比如&#xff1a;人的姓名、年龄、身高、体重…都是对象的属性。 车的品牌、型号、颜色、载重量...都是对象的属性。 对象属性既可以在类外面添加和获…
阅读更多...
最新文章

Copyright @ 2022~2023