本站消息

站长简介/公众号


站长简介:逗比程序员,理工宅男,前每日优鲜python全栈开发工程师,利用周末时间开发出本站,欢迎关注我的微信公众号:程序员总部,程序员的家,探索程序员的人生之路!分享IT最新技术,关注行业最新动向,让你永不落伍。了解同行们的工资,生活工作中的酸甜苦辣,谋求程序员的最终出路!

  价值13000svip视频教程,java大神匠心打造,零基础java开发工程师视频教程全套,基础+进阶+项目实战,包含课件和源码

  出租广告位,需要合作请联系站长


+关注
已关注

分类  

暂无分类

标签  

暂无标签

日期归档  

2021-05(16)

2021-06(58)

2021-07(11)

2021-08(50)

2021-09(37)

大数据平台实时数仓从0到1搭建之 - 09 Flink 安装配置

发布于2021-10-05 13:00     阅读(642)     评论(0)     点赞(10)     收藏(3)


概述

Flink

集群规划

server110server111server112
FlinkJobManager
TaskManager
TaskManagerTaskManager

安装配置

[root@server110 software]# tar -xzvf flink-1.13.2-bin-scala_2.11.tgz -C /opt/modules/
[root@server110 modules]# cd flink-1.13.2/conf/
#配置文件
[root@server110 conf]# vim flink-conf.yaml
jobmanager.rpc.address: server110
#配置jobManager
[root@server110 conf]# vim masters
server110:8081
#配置taskManager
[root@server110 conf]# vim workers
server110
server111
server112

文件同步

[root@server110 modules]# scp -r flink-1.13.2/ server111:/opt/modules/
[root@server110 modules]# scp -r flink-1.13.2/ server112:/opt/modules/

Standalone模式集群启动

[root@server110 flink-1.13.2]# bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host server110.
Starting taskexecutor daemon on host server110.
Starting taskexecutor daemon on host server111.
Starting taskexecutor daemon on host server112.
[root@server110 flink-1.13.2]# /opt/jpsall.sh 
----------------jps server110 --------------------
31744 QuorumPeerMain
34785 StandaloneSessionClusterEntrypoint
35108 TaskManagerRunner
32182 Kafka
17387 NameNode
17916 NodeManager
17549 DataNode
35215 Jps
----------------jps server111 --------------------
3504 TaskManagerRunner
17265 DataNode
17540 NodeManager
532 QuorumPeerMain
17383 ResourceManager
18029 JobHistoryServer
3582 Jps
975 Kafka
----------------jps server112 --------------------
26338 QuorumPeerMain
17207 DataNode
17289 SecondaryNameNode
26765 Kafka
29565 Jps
29486 TaskManagerRunner
17407 NodeManager

web UI

http://server110:8081/
Flink自带的web管理界面
三个节点正常启动

在这里插入图片描述
在这里插入图片描述

测试

web界面直接提交任务,
Entry Class : 有main方法的入口类名,带包名
Parallelism : 并行度

在这里插入图片描述
填写好对应的入口类,并行度为1
在这里插入图片描述
查看执行计划
在这里插入图片描述
提交任务
在这里插入图片描述
在TaskManager中看到,server110这个节点Free slots为0,说明任务在这个节点上运行
在这里插入图片描述
因为代码写的直接把结果输出到控制台,所以查看server110的stdout,可以看到对应的输出
在这里插入图片描述

关闭集群

[root@server110 flink-1.13.2]# bin/stop-cluster.sh 
Stopping taskexecutor daemon (pid: 39167) on host server110.
Stopping taskexecutor daemon (pid: 6299) on host server111.
Stopping taskexecutor daemon (pid: 31624) on host server112.
Stopping standalonesession daemon (pid: 38853) on host server110.

Flink on Yarn

用yarn管理flink任务,需要导一个jar包
flink-shaded-hadoop3-uber-blink-3.7.0.jar
这个jar包我是用maven下载的,直接下载不下来

<dependency>
    <groupId>com.alibaba.blink</groupId>
    <artifactId>flink-shaded-hadoop3-uber</artifactId>
    <version>blink-3.7.0</version>
</dependency>

同步到其他机器

[root@server110 flink-1.13.2]# scp lib/flink-shaded-hadoop3-uber-blink-3.7.0.jar server111:/opt/modules/flink-1.13.2/lib/
[root@server110 flink-1.13.2]# scp lib/flink-shaded-hadoop3-uber-blink-3.7.0.jar server112:/opt/modules/flink-1.13.2/lib/

提交任务

使用Flink on yarn 模式启动任务,可以直接在yarn管理界面查看到任务http://server111:8088/
点击右方的Tracking UI,可直接跳转到Flink管理界面,
在Running Jobs里可以看到当前正在运行的job,
在TaskManagers里可以看到当前运行的job的运行情况,因为代码直接写的打印,所以最终结果打印在Stdout 里
点击Cancel job,任务停止,刷新flink管理界面自动跳转到yarn管理界面,状态为killed

提交任务脚本,不需要启动start-cluster,直接执行flink run就行

[root@server110 flink-1.13.2]# bin/flink run -m yarn-cluster -c com.z.WordCount /opt/FlinkDemo-1.0-SNAPSHOT-jar-with-dependencies.jar

yarn管理界面
在这里插入图片描述

点击Tracking UI , 直接跳转到flink管理界面
在这里插入图片描述

查看TaskMamager,yarn管理,只启动需要的节点,所以只有一个节点正在运行
在这里插入图片描述
查看输出结果
在这里插入图片描述

取消任务
在这里插入图片描述

返回yarn管理界面,任务状态killed,Tracking UI也没了
在这里插入图片描述

测试结束

安装完成

附件:kafka生产者

需要手动输入数据,才能看到效果

[root@server111 kafka_2.11-2.4.1]# bin/kafka-console-producer.sh --broker-list server112:9092 --topic test-topic
>hello world
>hello sara
>hello scala
>hello flink
>

附件:WordCount 代码

<flink.version>1.13.2</flink.version>
<scala.version>2.11</scala.version>
package com.z

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

import java.util.Properties

/**
 * @author wenzheng.ma 
 * @date 2021-10-03 19:43
 * @desc
 */
object WordCount {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度1
    env.setParallelism(1)

    //topic
    val topic = "test-topic"
    //kafka的配置信息
    val prop = new Properties()
    prop.setProperty("bootstrap.servers", "server110:9092,server111:9092,server112:9092")
    prop.setProperty("group.id", "test-group")

    //创建kafka数据源
    val kafka = new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), prop)

    //添加kafka数据源
    val inputStream = env.addSource(kafka)
    val resultStream = inputStream
      .flatMap(_.split(" "))
      .map((_, 1))
      .keyBy(0)
      .sum(1)
    //打印结果
    resultStream.print()
    //阻塞进程,一直等待数据
    env.execute()
  }
}

原文链接:https://blog.csdn.net/sinat_25528181/article/details/120596220



所属网站分类: 技术文章 > 博客

作者:niceboty

链接:http://www.javaheidong.com/blog/article/297035/f2fdccf50a5d936e462f/

来源:java黑洞网

任何形式的转载都请注明出处,如有侵权 一经发现 必将追究其法律责任

10 0
收藏该文
已收藏

评论内容:(最多支持255个字符)