Apache Flink（五）：Apache Flink快速入门 - 环境准备及入门案例

news2025/2/26 13:20:05

🏡 个人主页：IT贫道_大数据OLAP体系技术栈,Apache Doris,Clickhouse 技术-CSDN博客

 🚩 私聊博主：加入大数据技术讨论群聊，获取更多大数据资料。

 🔔 博主个人B栈地址：豹哥教你大数据的个人空间-豹哥教你大数据个人主页-哔哩哔哩视频

目录

1. Flink开发环境准备

2. Flink入门案例

2.1 IDEA Project创建及配置

​​​​​​​2.2 案例数据准备

​​​​​​​1. Flink开发环境准备

学习一门新的编程语言时，往往会从“hello world”程序开始，而接触一套新的大数据计算框架时，则一般会从WordCount案例入手，下面以大数据中最经典入门案例WordCount为例，来编写Flink代码，Flink底层源码是基于Java代码进行开发，在Flink编程中我们除了可以使用Java语言来进行编写Flink程序外，还可以使用Scala、Python语言来进行编写Flink程序，在后续章节中我们将会主要使用Java和Scala来编写Flink程序。下面来准备下Flink开发环境。

  • Flink版本

本套课程中我们采用Flink最新版本1.16.0，Flink1.16.0版本官方文档地址：https://nightlies.apache.org/flink/flink-docs-release-1.16/

  • JDK环境

Flink核心模块均采用Java开发，所以运行环境需要依赖JDK,Flink可以基于类UNIX 环境中运行，例如：Linux、Max OS、Windows等，在这些系统上运行Flink时都需要配置JDK环境，Flink 1.16.0版本需要JDK版本为JDK11,目前版本也支持使用JDK8，后续版本对JDK8的支持将会移除。

考虑到Flink后期与一些大数据框架进行整合，这些大数据框架对JDK11的支持并不完善，例如：Hive3.1.3版本还不支持JDK11，所以本课程采用JDK8来开发Flink。对JDK8安装及配置不再详述。

附：JDK11 下载地址如下：

https://www.oracle.com/java/technologies/javase-jdk11-downloads.html。

  • 开发工具

我们可以选择IntelliJ IDEA或者Eclipse作为Flink应用的开发IDE，Flink开发官方建议使用IntelliJ IDEA，因为它默认集成了Scala和Maven环境，使用更加方便，我们这门课使用IntelliJ IDEA开发工具，具体安装步骤不再详述。

  • Maven环境

通过IntelliJ IDEA进行开发Flink Application时，可以使用Maven来作为项目jar包管理工具，需要在本地安装Maven及配置Maven的环境变量，需要注意的是，Maven版本需要使用3.0.4及以上，否则编译或开发过程中会有问题。这里使用Maven 3.2.5版本。

  • Scala环境

Flink开发语言可以选择Java、Scala、Python，如果用户选择使用Scala作为Flink应用开发语言，则需要安装Scala执行环境。

在Flink1.15之前版本，如果只是使用Flink的Java api ，对于一些没有Scala模块的包和表相关模块的包需要在Maven引入对应的包中加入scala后缀，例如：flink-table-planner_2.11，后缀2.11代表的就是Scala版本。在Flink1.15.0版本后，Flink添加对opting-out（排除） Scala的支持，如果你只使用Flink的Java api，导入包也不必包含scala后缀，你可以使用任何Scala版本。如果使用Flink的Scala api，需要选择匹配的Scala版本。

从Flink1.7版本往后支持Scala 2.11和2.12版本，目前Flink1.15.0版本后只支持Scala 2.12，不再支持Scala 2.11。Scala环境可以通过本地安装Scala执行环境，也可以通过Maven依赖Scala-lib引入，如果本地安装了Scala某个版本，建议在Maven中添加Scala-lib依赖。Scala2.12.8之后的版本与之前的2.12.x版本不兼容,建议使用Scala2.12.8之后版本。

  • Hadoop环境

Flink可以操作HDFS中的数据及基于Yarn进行资源调度，所以需要对应的Hadoop环境，Flink1.16.0版本支持的Hadoop最低版本为2.8.5，本课程中我们使用Hadoop3.3.4版本。关于Hadoop3.3.4版本搭建，参照第三章节。

​​​​​​​2. Flink入门案例

需求：读取本地数据文件，统计文件中每个单词出现的次数。​​​​​​​

2.1 IDEA Project创建及配置

本课程编写Flink代码选择语言为Java和Scala，所以这里我们通过IntelliJ IDEA创建一个目录，其中包括Java项目模块和Scala项目模块，将Flink Java api和Flink Scala api分别在不同项目模块中实现。步骤如下：

1) 打开IDEA，创建空项目

2) 在IntelliJ IDEA中安装Scala插件

使用IntelliJ IDEA开发Flink，如果使用Scala api 那么还需在IntelliJ IDEA中安装Scala的插件，如果已经安装可以忽略此步骤，下图为以安装Scala插件。

3) 打开Structure,创建项目新模块

创建Java模块：

继续点击“+”，创建Scala模块：

创建好“FlinkScalaCode”模块后，右键该模块添加Scala框架支持，并修改该模块中的“java”src源为“scala”:

在“FlinkScalaCode”模块Maven pom.xml中引入Scala依赖包，这里使用的Scala版本为2.12.10。

<!-- Scala包 -->
<dependency>
  <groupId>org.scala-lang</groupId>
  <artifactId>scala-library</artifactId>
  <version>2.12.10</version>
</dependency>
<dependency>
  <groupId>org.scala-lang</groupId>
  <artifactId>scala-compiler</artifactId>
  <version>2.12.10</version>
</dependency>
<dependency>
  <groupId>org.scala-lang</groupId>
  <artifactId>scala-reflect</artifactId>
  <version>2.12.10</version>
</dependency>

4) Log4j日志配置

为了方便查看项目运行过程中的日志，需要在两个项目模块中配置log4j.properties配置文件，并放在各自项目src/main/resources资源目录下，没有resources资源目录需要手动创建并设置成资源目录。log4j.properties配置文件内容如下：

log4j.rootLogger=ERROR, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss} %p %c{2}: %m%n

并在两个项目中的Maven pom.xml中添加对应的log4j需要的依赖包，使代码运行时能正常打印结果：

<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-log4j12</artifactId>
  <version>1.7.36</version>
</dependency>


<dependency>
  <groupId>org.apache.logging.log4j</groupId>
  <artifactId>log4j-to-slf4j</artifactId>
  <version>2.17.2</version>
</dependency>

5) 分别在两个项目模块中导入Flink Maven依赖

“FlinkJavaCode”模块导入Flink Maven依赖如下：

<properties>
  <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  <maven.compiler.source>1.8</maven.compiler.source>
  <maven.compiler.target>1.8</maven.compiler.target>
  <flink.version>1.16.0</flink.version>
  <slf4j.version>1.7.36</slf4j.version>
  <log4j.version>2.17.2</log4j.version>
</properties>

<dependencies>

  <!-- Flink DataStream 依赖包 -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
  </dependency>


  <!-- Flink批和流开发依赖包 -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
  </dependency>

  <!-- slf4j&log4j 日志相关包 -->
  <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>${slf4j.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-to-slf4j</artifactId>
    <version>${log4j.version}</version>
  </dependency>

</dependencies>

“FlinkScalaCode”模块导入Flink Maven依赖如下：

<properties>
  <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  <maven.compiler.source>1.8</maven.compiler.source>
  <maven.compiler.target>1.8</maven.compiler.target>
  <flink.version>1.16.0</flink.version>
  <slf4j.version>1.7.31</slf4j.version>
  <log4j.version>2.17.1</log4j.version>
  <scala.version>2.12.10</scala.version>
  <scala.binary.version>2.12</scala.binary.version>
</properties>

<dependencies>
  <!-- Flink批和流开发依赖包 -->
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-scala_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
  </dependency>


  <!-- Scala包 -->
  <dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>${scala.version}</version>
  </dependency>
  <dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-compiler</artifactId>
    <version>${scala.version}</version>
  </dependency>
  <dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-reflect</artifactId>
    <version>${scala.version}</version>
  </dependency>

  <!-- slf4j&log4j 日志相关包 -->
  <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>${slf4j.version}</version>
  </dependency>
  <dependency>
    <groupId>org.apache.logging.log4j</groupId>
    <artifactId>log4j-to-slf4j</artifactId>
    <version>${log4j.version}</version>
  </dependency>

</dependencies>

注意：在后续实现WordCount需求时，Flink Java Api只需要在Maven中导入“flink-clients”依赖包即可，而Flink Scala Api 需要导入以下三个依赖包：

flink-scala_${scala.binary.version}

flink-streaming-scala_${scala.binary.version}

flink-clients

主要是因为在Flink1.15版本后，Flink添加对opting-out（排除）Scala的支持，如果你只使用Flink的Java api，导入包不必包含scala后缀，如果使用Flink的Scala api，需要选择匹配的Scala版本。

​​​​​​​​​​​​​​2.2 案例数据准备

在项目“MyFlinkCode”中创建“data”目录，在目录中创建“words.txt”文件，向文件中写入以下内容，方便后续使用Flink编写WordCount实现代码。

hello Flink
hello MapReduce
hello Spark
hello Flink
hello Flink
hello Flink
hello Flink
hello Java
hello Scala
hello Flink
hello Java
hello Flink
hello Scala
hello Flink
hello Flink

 更多Flink内容参考下个博文：Apache Flink（六）：Apache Flink快速入门 - Flink案例实现

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

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

相关文章

6-55.汽车类的继承

6-55.汽车类的继承

根据给定的汽车类vehicle&#xff08;包含的数据成员有车轮个数wheels和车重weight&#xff09;声明&#xff0c;完成其中成员函数的定义&#xff0c;之后再定义其派生类并完成测试。 小车类car是它的派生类&#xff0c;其中包含载人数passenger_load。每个类都有相关数据的输出…
阅读更多...
AD23等间距拉线、布线的方法

AD23等间距拉线、布线的方法

U M 键进行多根走线&#xff0c; 多根走线想保持10个mil 我可以直接按table键,弹出Multi-Routing ponent&#xff0c;项的Bus Spadng输入框中填充10个mil&#xff0c;新走线产生10个mil的等间距 保持最小的一个规则&#xff0c;可以去到6mil线距。 在拉线操作过程中&#…
阅读更多...
HuggingFace学习笔记--BitFit高效微调

HuggingFace学习笔记--BitFit高效微调

1--BitFit高效微调 BitFit&#xff0c;全称是 bias-term fine-tuning&#xff0c;其高效微调只去微调带有 bias 的参数&#xff0c;其余参数全部固定&#xff1b; 2--实例代码 from datasets import load_from_disk from transformers import AutoTokenizer, AutoModelForCaus…
阅读更多...
牛客算法题【HJ96 表示数字】golang实现

牛客算法题【HJ96 表示数字】golang实现

题目 HJ96 表示数字 golang实现 package mainimport ("fmt""unicode" )func main() {s : ""var s_o stringvar char_pre, r runefor {n, _ : fmt.Scan(&s)if n 0 {break} else {for _, r range s {if unicode.IsDigit(r) {if !unicode.…
阅读更多...
Kubernetes技术与架构-策略

Kubernetes技术与架构-策略

Kubernetes集群提供系统支持的策略&#xff0c;也提供开放接口给第三方定义的策略&#xff0c;这些策略用于可定义的配置文件或者Kubernetes集群的运行时环境&#xff0c;其中包括进程ID数量的申请与限制策略&#xff0c;服务器节点Node内的进程ID的数量限制策略&#xff0c;Po…
阅读更多...
MySQL 临时数据空间不足导致SQL被killed 的问题与扩展

MySQL 临时数据空间不足导致SQL被killed 的问题与扩展

开头还是介绍一下群&#xff0c;如果感兴趣PolarDB ,MongoDB ,MySQL ,PostgreSQL ,Redis, Oceanbase, Sql Server等有问题&#xff0c;有需求都可以加群群内&#xff0c;可以解决你的问题。加群请联系 liuaustin3 &#xff0c;&#xff08;共1730人左右 1 2 3 4 5&#xff0…
阅读更多...
网站有必要使用SSL证书吗

网站有必要使用SSL证书吗

随着互联网的快速发展&#xff0c;网络安全问题也变得日益突出&#xff0c;SSL证书的作用日益凸显。 什么是SSL证书&#xff1f; SSL证书&#xff08;Secure Sockets Layer Certificate&#xff09;&#xff0c;也称为TLS证书&#xff08;Transport Layer Security Certifica…
阅读更多...
【论文阅读】CAN网络中基于时序信道的隐蔽认证算法

【论文阅读】CAN网络中基于时序信道的隐蔽认证算法

文章目录 摘要一、引言和动机A 相关工作 二、背景及实验设置A 以前工作中的时钟偏差和局限性B.最坏到达时间C.安装组件 三、优化流量分配A.问题陈述B.优化帧调度 四、协议和结果A.主协议B.对手模型C. 优化流量和单一发送者的结果D.多发送方情况和噪声信道E.信道数据速率&#x…
阅读更多...
HuggingFace学习笔记--Prompt-Tuning高效微调

HuggingFace学习笔记--Prompt-Tuning高效微调

1--Prompt-Tuning介绍 Prompt-Tuning 高效微调只会训练新增的Prompt的表示层&#xff0c;模型的其余参数全部固定&#xff1b; 新增的 Prompt 内容可以分为 Hard Prompt 和 Soft Prompt 两类&#xff1b; Soft prompt 通常指的是一种较为宽泛或模糊的提示&#xff0c;允许模型在…
阅读更多...
ELK配置记录

ELK配置记录

1. filebeat.yml配置 启动命令&#xff1a; ./filebeat -e -c filebeat.yml # 输入 filebeat.inputs: - type: logenabled: truepaths:- /soft/log/base.*#跨行日志正则&#xff0c;从有时间的开始&#xff0c;到下一个时间之前结束multiline.pattern: ^\[[0-9]{4}-[0-9]{2}…
阅读更多...
Concurrent Security of Anonymous Credentials Light, Revisited

Concurrent Security of Anonymous Credentials Light, Revisited

目录 摘要引言 摘要 我们重新审视了著名的匿名证书轻&#xff08;ACL&#xff09;方案&#xff08;Baldimtsi和Lysyanskaya&#xff0c;CCS’13&#xff09;的并发安全保证。该方案最初被证明在按顺序执行时是安全的&#xff0c;其并发安全性仍然是一个悬而未决的问题。Benham…
阅读更多...
【PyTorch】线性回归

【PyTorch】线性回归

文章目录 1. 代码实现1.1 一元线性回归模型的训练 2. 代码解读2.1. tensorboardX2.1.1. tensorboardX的安装2.1.2. tensorboardX的使用 1. 代码实现 波士顿房价数据集下载 1.1 一元线性回归模型的训练 import numpy as np import torch import torch.nn as nn from torch.ut…
阅读更多...
[计算机网络] 高手常用的几个抓包工具（上）

[计算机网络] 高手常用的几个抓包工具（上）

文章目录 高手常用的抓包工具一览什么是抓包工具优秀抓包工具WiresharkFiddlerTcpdumpCharles 高手常用的抓包工具一览 什么是抓包工具 抓包工具是一种可以捕获、分析和修改网络流量的软件。它可以帮助您进行网络调试、性能测试、安全审计等任务。 抓包工具可以实时地显示网…
阅读更多...
使用trigger-forward跨流水线传递参数

使用trigger-forward跨流水线传递参数

参考文档&#xff1a;https://docs.gitlab.com/ee/ci/yaml/#triggerforward 今天给大家介绍一个gitlab CI/CD的关键字 - forward&#xff0c;该关键字是一个比较偏的功能&#xff0c;但同时也是一个很实用的功能&#xff0c;我们通过在gitlab的ci文件中使用forward关键字&#…
阅读更多...
Android HCI日志分析案例1

Android HCI日志分析案例1

案例1--蓝牙扫描设备过程分析 应用层发起搜索蓝牙设备&#xff0c;Android 官方提供的蓝牙扫描方式有三种&#xff0c;分别如下&#xff1a; BluetoothAdapter.startDiscovery(); //可以扫描经典蓝牙和BLE两种。BluetoothAdapter.startLeScan();//扫描低功耗蓝牙&#xff0c;…
阅读更多...
深入理解同源限制：网络安全的守护者(上)

深入理解同源限制：网络安全的守护者(上)

&#x1f90d; 前端开发工程师&#xff08;主业&#xff09;、技术博主&#xff08;副业&#xff09;、已过CET6 &#x1f368; 阿珊和她的猫_CSDN个人主页 &#x1f560; 牛客高级专题作者、在牛客打造高质量专栏《前端面试必备》 &#x1f35a; 蓝桥云课签约作者、已在蓝桥云…
阅读更多...
超简单的node脚本，将xlsx文件转化为json

超简单的node脚本，将xlsx文件转化为json

开发场景&#xff0c;在一个官网中&#xff0c;官网的设计非常简单&#xff0c;就是一个纯静态的页面&#xff0c;全网站仅一个地方调一下接口&#xff0c;发一下用户填写的信息到运营同学的邮箱&#xff0c;这些数据不会记录在数据库&#xff0c;我需要做一个这样的下拉框。 但…
阅读更多...
python使用记录

python使用记录

1、VSCode添加多个python解释器 只需要将对应的python.exe的目录&#xff0c;添加到系统环境变量的Path中即可&#xff0c;VSCode会自动识别及添加 2、pip 使用 pip常用命令和一些坑 查看已安装库的版本号 pip show 库名称 通过git 仓库安装第三方库 pip install git仓库地…
阅读更多...
SQL Server 2016（分离和附加数据库）

SQL Server 2016（分离和附加数据库）

1、实验环境。 基于上一个实验《SQL Server&#xff08;创建数据库&#xff09;》 2、需求描述。 class数据库的数据文件和事务日志文件都位于C:\db_class目录下。现在需要把class数据库的数据文件和事务日志文件分开存放&#xff0c;数据文件class.mdf存放于原位置&#xff0…
阅读更多...
DQN原理及PyTorch实现【强化学习】

DQN原理及PyTorch实现【强化学习】

NSDT工具推荐&#xff1a; Three.js AI纹理开发包 - YOLO合成数据生成器 - GLTF/GLB在线编辑 - 3D模型格式在线转换 - 可编程3D场景编辑器 - REVIT导出3D模型插件 - 3D模型语义搜索引擎 欢迎来到我们的强化学习系列的第三部分。 在上两篇博客中&#xff0c;我们介绍了强化学习中…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023