Flink-01学习 介绍Flink及上手小项目之词频统计

news2025/4/23 2:54:16

flink简介

官网
概述：
学习Flink具体包括四个关键概念：流数据的持续处理，事件时间，有状态流处理和状态快照。
Apache Flink 是一个开源的流处理框架，旨在处理批处理和实时数据处理，具有高吞吐量和低延迟的特点。
Apache Flink 是一个用于对无界和有界数据流进行有状态计算的框架和分布式处理引擎。Flink 被设计为可在所有常见的集群环境中运行，并以内存速度和任意规模执行计算。
特点：

  • 流处理：Flink 将批处理视为流处理的一种特殊情况。这种方法允许实时数据处理，实现即时的洞察和行动。
  • 有状态计算：Flink 提供强大的状态管理，使得在处理流的过程中可以保持状态。这一特性对于需要容错和一致性的应用至关重要。
  • 事件时间处理：Flink 允许用户基于事件时间来处理数据，即使数据无序到达，也能提供准确及时的结果。
  • 容错性：Flink 的状态管理和检查点机制确保系统在出现故障时能够恢复而不丢失状态，维护数据完整性和应用一致性。
  • 高吞吐量和低延迟：Flink 的架构优化了高吞吐量和低延迟，适合高性能应用。
  • 可扩展性：Flink 可以扩展到数千个节点，能够处理大规模数据处理任务。
  • 灵活的部署选项：Flink 可以部署在各种环境中，包括独立集群、云环境和容器编排平台（如 Kubernetes）。
    应用：
    Flink 集群始终由一个 JobManager和一个或多个 Flink TaskManager组成。JobManager 负责处理作业提交、作业监管以及资源管理。Flink TaskManager 是工作进程，负责执行构成 Flink 作业的实际 任务。在本实践中，您将从单个 TaskManager 开始，但稍后可以扩展到更多 TaskManager。

实践

创建一个meven项目
引入pom

  <?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>org.example</groupId>
    <artifactId>flink-demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.20.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>1.20.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.20.0</version>
        </dependency>


        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java-bridge -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge</artifactId>
            <version>1.20.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table-api-java -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java</artifactId>
            <version>1.20.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>1.20.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.maven</groupId>
            <artifactId>maven-plugin-api</artifactId>
            <version>2.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.maven.plugin-tools</groupId>
            <artifactId>maven-plugin-annotations</artifactId>
            <version>3.2</version>
        </dependency>
        <dependency>
            <groupId>org.codehaus.plexus</groupId>
            <artifactId>plexus-utils</artifactId>
            <version>3.0.8</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.8.2</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-plugin-plugin</artifactId>
            <version>3.2</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <configuration>
                        <filters>
                            <filter>
                                <artifact>*:*</artifact>
                                <excludes>
                                    <exclude>META-INF/*.SF</exclude>
                                    <exclude>META-INF/*.DSA</exclude>
                                    <exclude>META-INF/*.RSA</exclude>
                                </excludes>
                            </filter>
                        </filters>
                    </configuration>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <configuration>
                <source>8</source>
                <target>8</target>
            </configuration>
        </plugin>
    </plugins>
</build>
</project>

Flink 中的 DataStream 程序是常规程序，用于对数据流进行转换（例如，过滤、更新状态、定义窗口、聚合）。数据流最初由各种来源（例如，消息队列、套接字流、文件）创建。结果通过接收器返回，接收器可以将数据写入文件或标准输出（例如命令行终端）。Flink 程序可以在各种上下文中运行，独立运行或嵌入到其他程序中。执行可以在本地 JVM 中，也可以在多台机器组成的集群中执行。

代码如下：

package org.example.snow.demo1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.datastream.api.ExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author snowsong
 */
public class StartRunApp {
    public static void main(String[] args) throws Exception {
        // 数据来源
        String inPath = "dataMsg/dataSourceFile.txt";
        // 数据输出
        String outputPath = "dataMsg/result.csv";
        // 初始化
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        // 读取文件信息
        DataStreamSource<String> textFile = executionEnvironment.readTextFile(inPath);
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = textFile
                .flatMap(new Splitter())
                .keyBy(value -> value.f0)
                .sum(1);
        textFile.print();
        // 写入数据到 CSV 文件，使用 writeAsText 或 writeAsCsv
        flatMap.writeAsCsv(outputPath,FileSystem.WriteMode.OVERWRITE," ","\n")
                // 这是设置并行度的参数，表示执行该操作的任务的数量。在这里，setParallelism(1) 表示该操作会在 单个任务 中执行，而不是并行执行多个任务。通常情况下，Flink 在处理流或批数据时可以通过并行执行来加速处理，而设置并行度为 1 可以强制数据写入在一个线程中进行。
                .setParallelism(1);
        // 调用 execute() 后，Flink 会开始处理数据流中的每个操作，并根据定义的逻辑执行数据转换、聚合等操作。
        // 作业名称："file.txt -> result.csv" 是为了描述输入文件和输出文件的关系，可以帮助你理解这个作业的目的。这个名称在 Flink 的执行日志和监控界面中会显示
        executionEnvironment.execute("file.txt -> result.csv");

    }

    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }
}

运行结果 将文章进行了分词处理
请添加图片描述

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

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

相关文章

目标检测篇---R-CNN梳理

目标检测篇---R-CNN梳理

目标检测系列文章 第一章 R-CNN 目录 目标检测系列文章&#x1f4c4; 论文标题&#x1f9e0; 论文逻辑梳理1. 引言部分梳理 (动机与思想) &#x1f4dd; 三句话总结&#x1f50d; 方法逻辑梳理&#x1f680; 关键创新点&#x1f517; 方法流程图补充边界框回归 (BBR)1. BBR 的…
阅读更多...
C#处理网络传输中不完整的数据流

C#处理网络传输中不完整的数据流

1、背景 在读取byte数组的场景&#xff08;例如&#xff1a;读取文件、网络传输数据&#xff09;中&#xff0c;特别是网络传输的场景中&#xff0c;非常有可能接收了不完整的byte数组&#xff0c;在将byte数组转换时&#xff0c;因字符的缺失/增多&#xff0c;转为乱码。如下…
阅读更多...
HTML 初识

HTML 初识

段落标签 <p><!-- 段落标签 -->Lorem ipsum dolor sit amet consectetur adipisicing elit. Fugiat, voluptate iure. Obcaecati explicabo sint ipsum impedit! Dolorum omnis voluptas sint unde sed, ipsa molestiae quo sapiente quos et ad reprehenderit.&l…
阅读更多...
MATLAB 训练CNN模型 yolo v4

MATLAB 训练CNN模型 yolo v4

学生对小车控制提出了更好的要求&#xff0c;能否加入深度学习模型。 考虑到小车用matlab来做&#xff0c;yolo v5及以上版本都需要在pytorch下训练&#xff0c;还是用早期版本来演示。 1 yolov4 调用 参考 trainYOLOv4ObjectDetector (mathworks.com) name "tiny-yo…
阅读更多...
【前端】跟着maxkb学习logicflow流程图画法

【前端】跟着maxkb学习logicflow流程图画法

文章目录 背景1. 选定学习对象-maxkb应用逻辑编排2. 确定实现框架3. 关键逻辑&#xff1a;查看app-node.js4. 学习开始节点绘制流程数据形式 5. 给节点增加表单输入框遇到过的问题 背景 看看前端如何绘制流程图&#xff0c;界面好看点。 "logicflow/core": "1.…
阅读更多...
【漏洞复现】CVE-2024-38856（ApacheOfbiz RCE）

【漏洞复现】CVE-2024-38856（ApacheOfbiz RCE）

【漏洞复现】CVE-2024-38856&#xff08;ApacheOfbiz RCE&#xff09; 1. 漏洞描述 Apache OFBiz 是一个开源的企业资源规划&#xff08;ERP&#xff09;系统。它提供了一套企业应用程序&#xff0c;用于集成和自动化企业的许多业务流程。 这个漏洞是由于对 CVE-2023-51467 的…
阅读更多...
超详细VMware虚拟机扩容磁盘容量-无坑版

超详细VMware虚拟机扩容磁盘容量-无坑版

1.环境&#xff1a; 虚拟机&#xff1a;VMware Workstation 17 Pro-17.5.2 Linux系统&#xff1a;Ubuntu 22.04 LTS 2.硬盘容量 虚拟机当前硬盘容量180G -> 扩展至 300G 3.操作步骤 &#xff08;1&#xff09;在虚拟机关机的状态下&#xff0c;虚拟机硬盘扩容之前必…
阅读更多...
全面理解Linux 系统日志：核心文件与查看方法

全面理解Linux 系统日志：核心文件与查看方法

全文目录 1 Linux 系统日志分类及功能1.1 通用日志1.1.1 ‌/var/log/messages1.1.2 ‌/var/log/syslog 1.2 安全相关日志1.2.1 ‌/var/log/auth.log‌&#xff08;Debian/Ubuntu&#xff09;或 ‌/var/log/secure‌&#xff08;RHEL/CentOS&#xff09;1.2.2 /var/log/audit/au…
阅读更多...
机器学习-08-关联规则更新

机器学习-08-关联规则更新

总结 本系列是机器学习课程的系列课程&#xff0c;主要介绍机器学习中关联规则和协同过滤。 参考 机器学习&#xff08;三&#xff09;&#xff1a;Apriori算法&#xff08;算法精讲&#xff09; Apriori 算法 理论 重点 【手撕算法】【Apriori】关联规则Apriori原理、代码…
阅读更多...
Flutter与FastAPI的OSS系统实现

Flutter与FastAPI的OSS系统实现

作者&#xff1a;孙嘉成 目录 一、对象存储 二、FastAPI与对象存储 2.1 缤纷云S4服务API对接与鉴权实现 2.2 RESTful接口设计与异步路由优化 三、Flutter界面与数据交互开发 3.1 应用的创建 3.2页面的搭建 3.3 文件的上传 关键词&#xff1a;对象存储、FastAPI、Flutte…
阅读更多...
Kubernetes控制平面组件：API Server详解（二）

Kubernetes控制平面组件：API Server详解（二）

云原生学习路线导航页&#xff08;持续更新中&#xff09; kubernetes学习系列快捷链接 Kubernetes架构原则和对象设计&#xff08;一&#xff09;Kubernetes架构原则和对象设计&#xff08;二&#xff09;Kubernetes架构原则和对象设计&#xff08;三&#xff09;Kubernetes控…
阅读更多...
MySQL-锁机制3-意向共享锁与意向排它锁、死锁

MySQL-锁机制3-意向共享锁与意向排它锁、死锁

文章目录 一、意向锁二、死锁应该如何避免死锁问题&#xff1f; 总结 一、意向锁 在表获取共享锁或者排它锁时&#xff0c;需要先检查该表有没有被其它事务获取过X锁&#xff0c;通过意向锁可以避免大量的行锁扫描&#xff0c;提升表获取锁的效率。意向锁是一种表级锁&#xf…
阅读更多...
报告系统状态的连续日期 mysql + pandas(连续值判断)

报告系统状态的连续日期 mysql + pandas(连续值判断)

本题用到知识点&#xff1a;row_number(), union, date_sub(), to_timedelta()…… 目录 思路 pandas Mysql 思路 链接&#xff1a;报告系统状态的连续日期 思路&#xff1a; 判断连续性常用的一个方法&#xff0c;增量相同的两个列的差值是固定的。 让日期与行号 * 天数…
阅读更多...
Tailwind 武林奇谈：bg-blue-400 失效，如何重拾蓝衣神功？

Tailwind 武林奇谈：bg-blue-400 失效，如何重拾蓝衣神功？

前言 江湖有云，Tailwind CSS，乃前端武林中的轻功秘籍。习得此技，排版如行云流水，配色似御风随形，收放自如，随心所欲。 某日，小侠你奋笔敲码，正欲施展“蓝衣神功”（bg-blue-400），让按钮怒气冲冠、蓝光满面，怎料一招使出，画面竟一片白茫茫大地真干净，毫无半点杀气…
阅读更多...
开始放飞之先搞个VSCode

开始放飞之先搞个VSCode

文章目录 开始放飞之先搞个VSCode重要提醒安装VSCode下载MinGW-w64回到VSCode中去VSCode原生调试键盘问题遗留问题参考文献 开始放飞之先搞个VSCode 突然发现自己的新台式机上面连个像样的编程环境都没有&#xff0c;全是游戏了&#xff01;&#xff01;&#xff01;&#xff…
阅读更多...
基于SA模拟退火算法的车间调度优化matlab仿真,输出甘特图和优化收敛曲线

基于SA模拟退火算法的车间调度优化matlab仿真,输出甘特图和优化收敛曲线

目录 1.程序功能描述 2.测试软件版本以及运行结果展示 3.核心程序 4.本算法原理 5.完整程序 1.程序功能描述 基于SA模拟退火算法的车间调度优化matlab仿真,输出甘特图和优化收敛曲线。输出指标包括最小平均流动时间&#xff0c;最大完工时间&#xff0c;最小间隙时间。 2…
阅读更多...
【仿Mudou库one thread per loop式并发服务器实现】SERVER服务器模块实现

【仿Mudou库one thread per loop式并发服务器实现】SERVER服务器模块实现

SERVER服务器模块实现 1. Buffer模块2. Socket模块3. Channel模块4. Poller模块5. EventLoop模块5.1 TimerQueue模块5.2 TimeWheel整合到EventLoop5.1 EventLoop与线程结合5.2 EventLoop线程池 6. Connection模块7. Acceptor模块8. TcpServer模块 1. Buffer模块 Buffer模块&…
阅读更多...
uniapp h5接入地图选点组件

uniapp h5接入地图选点组件

uniapp h5接入地图选点组件 1、申请腾讯地图key2、代码接入2.1入口页面 &#xff08;pages/map/map&#xff09;templatescript 2.2选点页面&#xff08;pages/map/mapselect/mapselect&#xff09;templatescript 该内容只针对uniapp 打包h5接入地图选点组件做详细说明&#x…
阅读更多...
【随缘更新，免积分下载】Selenium chromedriver驱动下载（最新版135.0.7049.42）

【随缘更新，免积分下载】Selenium chromedriver驱动下载（最新版135.0.7049.42）

目录 一、chromedriver概述 二、chromedriver使用方式 三、chromedriver新版本下载&#x1f525;&#x1f525;&#x1f525; 四、Selenium与Chrome参数设置&#x1f525;&#x1f525; 五、Selenium直接操控已打开的Chrome浏览器&#x1f525;&#x1f525;&#x1f525;…
阅读更多...
jenkins批量复制Job项目的shell脚本实现

jenkins批量复制Job项目的shell脚本实现

背景 现在需要将“测试” 目录中的所有job全部复制到 一个新目录中 test2。可以结合jenkins提供的apilinux shell 进行实现。 测试目录的实际文件夹名称是 test。 脚本运行效果如下&#xff1a; [qdevsom5f-dev-hhyl shekk]$ ./copy_jenkins_job.sh 创建文件夹 test2 获取源…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023