devbox
寸_铁
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Docker下的Flink_docker flink

作者：寸_铁 | 2024-07-17 08:50:55

赞

踩

docker flink

        在之前的文章Docker下的Storm中讲了通过Storm来进行实时计算任务，那么这本篇文章中将介绍怎样通过Flink来做实时计算任务。

Storm和Flink都是流处理框架，用于处理实时数据流。它们在一些方面有一些异同点。

1.Storm和Flink的架构不同。Storm是一个分布式的、实时的、容错的流处理系统，采用了主从架构。它使用了一个中心的协调节点（Nimbus）来管理和分发作业，以及多个工作节点（Supervisor）来执行计算任务。而Flink是一个基于事件驱动的流处理框架，采用了分布式流处理引擎的架构。它使用了一个中央调度器（JobManager）来协调和管理作业，以及多个任务管理器（TaskManager）来执行计算任务。

2.Storm和Flink在容错性上有一些区别。Storm使用了可靠性机制来确保数据的完整性和一致性，例如元组的可靠性传递和消息确认机制。而Flink采用了基于检查点的容错机制，通过定期保存计算状态的检查点来实现故障恢复和容错。

3.Storm和Flink的处理模型也有所不同。Storm采用了基于元组的处理模型，将数据流抽象为无界的元组流，通过定制的Bolt进行处理。而Flink采用了基于数据流的处理模型，将数据流抽象为有界的数据流，通过转换操作和窗口操作进行处理。

4.Storm和Flink在一些特性和用途上也有一些差异。Storm适用于一些低延迟、高吞吐量的实时应用，如实时分析、实时推荐等。而Flink更加通用，适用于各种批处理和流处理的场景，支持复杂的流处理和批处理操作，并提供了高级API和丰富的库和工具集。

拉取Flink镜像

        在这里我们拉取的是1.13.0版本的Flink的官方Docker镜像：

docker pull flink:1.13.0

        拉取完成，通过命令进行检查，如果是Docker Desktop的话也可以直接在UI中检查镜像的拉取情况。

docker images

创建容器

        这里通过docker-compose定义一个包含一个作业管理器和两个任务管理器的 Apache Flink 集群。作业管理器映射到主机的 8081 端口，而任务管理器则在同一网络下运行，并设置了各自的环境变量以连接到作业管理器。

  1. version: "2"
  2. services:
  3.   jobmanager:
  4.     image: flink:1.13.0
  5.     ports:
  6.       - "8081:8081"
  7.     command: jobmanager
  8.     environment:
  9.       - |
  10.         FLINK_PROPERTIES=
  11.         jobmanager.rpc.address: jobmanager
  12.     networks:
  13.       - flink-net
  14.  
  15.   taskmanager-1:
  16.     image: flink:1.13.0
  17.     container_name: taskmanager-1
  18.     depends_on:
  19.       - jobmanager
  20.     command: taskmanager
  21.     environment:
  22.       - |
  23.         FLINK_PROPERTIES=
  24.         jobmanager.rpc.address: jobmanager
  25.         taskmanager.numberOfTaskSlots: 2
  26.     networks:
  27.       - flink-net
  28.   
  29.   taskmanager-2:
  30.     image: flink:1.13.0
  31.     container_name: taskmanager-2
  32.     depends_on:
  33.       - jobmanager
  34.     command: taskmanager
  35.     environment:
  36.       - |
  37.         FLINK_PROPERTIES=
  38.         jobmanager.rpc.address: jobmanager
  39.         taskmanager.numberOfTaskSlots: 2
  40.     networks:
  41.       - flink-net
  42.  
  43. networks:
  44.   flink-net:
  45.     driver: bridge

        在docker-compose.yml文件路径下的控制台输入下面的命令搭建集群：

docker-compose up

        当然也可以通过docker命令查看当前已经搭建好了的flink集群：

docker ps

集群控制页面(Flink UI)

        可以看到，在这里我们暴露了jobmanager的8081端口，通过这个端口我们可以打开Flink集群的UI界面并在上面提交程序jar包。

        通过在浏览器的地址栏中输入localhost:8081即可打开Flink的UI页面：

        Flink的UI页面是用于展示和监控Flink集群的运行状态和统计信息的Web可视化界面，通过它可以方便地查看集群总览、作业列表、作业监控、Task监控和日志查看等功能，帮助用户监控集群运行情况、发现和解决问题，并进行作业调优。

创建Flink任务程序

        Flink任务程序是基于Apache Flink框架编写的数据处理应用程序，它由一系列数据转换和操作组成，用于实现对数据流或批量数据的实时处理。任务程序可以定义数据源、数据转换操作和数据汇，同时可以设置数据窗口、触发器和窗口函数等，以实现各种复杂的数据处理逻辑。Flink任务程序具有高度可扩展性和容错性，能够处理大规模数据流和批量数据，并具备低延迟和高吞吐量的特性。

        在这里我们通过下面的实例来学习Flink项目的创建和任务的编写，在这一部分可能对一些特别深入细致的代码内容没办法进行详细的解释，如果读者想要对Flink进行深入的理解请学习Flink的系列课程，本篇文章仅仅带大家熟悉怎样在Docker中创建Flink集群并完成相应的任务。

        任务：现有游戏交易数据，其数据格式为(交易类型，交易金额)，现在请编写Flink Java任务来完成对不同交易类型的交易总金额进行统计。

Maven依赖导入

        如果要通过Flink向Redis,HBase等Sink数据需要导其他的依赖，这里我们只是将最终数据sink到控制台上，所以就不需要这些其他的依赖了，只需要最基础的Flink-Java依赖即可。

        由于flink依赖的包比较多，所以通过一个maven pom文件属性对这些依赖的作用范围进行管理：

  1. <properties>
  2.     <base.flink.scope>compile</base.flink.scope>
  3. </properties>

        再次基础上将依赖写入：

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.apache.flink</groupId>
  4.         <artifactId>flink-java</artifactId>
  5.         <version>1.13.0</version>
  6.         <scope>${base.flink.scope}</scope>
  7.     </dependency>
  8.     <dependency>
  9.         <groupId>org.apache.flink</groupId>
  10.         <artifactId>flink-streaming-java_2.12</artifactId>
  11.         <version>1.13.0</version>
  12.         <scope>${base.flink.scope}</scope>
  13.     </dependency>
  14.     <dependency>
  15.         <groupId>org.apache.flink</groupId>
  16.         <artifactId>flink-clients_2.12</artifactId>
  17.         <version>1.13.0</version>
  18.         <scope>${base.flink.scope}</scope>
  19.     </dependency>
  20. </dependencies>

编写自定义数据源

        Flink中已经实现好了的数据源包括本地集合、文件、Kafka、Socket等。本地集合数据源适用于小规模数据集的测试和调试，可以直接将数据加载到内存中进行处理。文件数据源可以从本地文件系统或分布式文件系统（如HDFS）中读取数据，支持各种格式的文件，如文本文件、CSV文件等。Kafka数据源可以将数据从Kafka消息队列中读取，实现实时数据处理。Socket数据源可以从网络socket中读取数据，常用于接收实时数据流。此外，Flink还支持自定义数据源，开发者可以根据自己的需求实现自己的数据源逻辑。这些数据源提供了灵活且高效的数据输入方式，使得用户可以方便地将各种数据接入到Flink中进行处理。

        虽然Flink中有许多已经实现好了的数据源，但是为了对Flink的自定义源进行了解，所以在这里我会自己手动实现一个自定义数据源。

        根据我们上面提到的任务，在这里我会创建一个自定义源来自动创建随机的交易流水：

  1. package source;
  2.  
  3. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  4.  
  5. import java.util.Random;
  6.  
  7. // 自定义实现Flink的SourceFunction接口，产生随机交易流数据
  8. public class RandomFlowSource implements SourceFunction<String> {
  9.  
  10.     // 创建一个Random对象用于生成随机数
  11.     private static final Random rd = new Random();
  12.  
  13.     // 用于控制数据源是否在运行的标志位
  14.     private volatile boolean isRunning = true;
  15.  
  16.  
  17.     @Override
  18.     public void run(SourceContext<String> sourceContext) throws Exception {
  19.         // 当数据源在运行时持续产生数据
  20.         while (isRunning) {
  21.             // 生成随机的交易类型，假设有3种交易类型
  22.             int transactionType = rd.nextInt(3);
  23.  
  24.             // 生成随机的交易金额，取值范围为0到100
  25.             double transactionAmount = rd.nextDouble() * 100;
  26.  
  27.             // 将交易类型和金额拼接成字符串，并将其发送给下游算子
  28.             sourceContext.collect(String.format("%d,%f", transactionType, transactionAmount));
  29.  
  30.             // 线程休眠100毫秒，模拟实际数据产生的时间间隔
  31.             Thread.sleep(100);
  32.         }
  33.     }
  34.  
  35.     @Override
  36.     public void cancel() {
  37.         // 当取消数据源时，修改标志位使数据源停止产生数据
  38.         isRunning = false;
  39.     }
  40. }

创建任务处理程序

        在完成了Flink数据源的创建之后，接下来就要完成最核心的任务处理部分的代码了。

  1. package tasks;
  2.  
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.api.java.functions.KeySelector;
  5. import org.apache.flink.api.java.tuple.Tuple2;
  6. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  7. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import scala.Int;
  10. import source.RandomFlowSource;
  11.  
  12. public class TransactionAnalysisJob {
  13.     public static void main(String[] args) throws Exception {
  14.         // 获取Flink的执行环境
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.  
  17.         // 添加自定义数据源RandomFlowSource作为数据输入源
  18.         DataStreamSource<String> dataSource = env.addSource(new RandomFlowSource());
  19.  
  20.         // 使用MapFunction将输入的字符串转换成Tuple2<String, Double>类型的数据流
  21.         SingleOutputStreamOperator<Tuple2<String, Double>> mapOp = dataSource.map(new MapFunction<String, Tuple2<String, Double>>() {
  22.             @Override
  23.             public Tuple2<String, Double> map(String s) throws Exception {
  24.                 // 将输入的字符串按逗号分割成交易类型和金额，并封装成Tuple2返回
  25.                 String[] fields = s.split(",");
  26.                 return Tuple2.of(fields[0], Double.parseDouble(fields[1]));
  27.             }
  28.         });
  29.  
  30.         // 按照交易类型进行分组，然后对交易金额进行求和
  31.         SingleOutputStreamOperator<Tuple2<String, Double>> sum = mapOp.keyBy(new KeySelector<Tuple2<String, Double>, String>() {
  32.             @Override
  33.             public String getKey(Tuple2<String, Double> stringDoubleTuple2) throws Exception {
  34.                 // 以交易类型作为key进行分组
  35.                 return stringDoubleTuple2.f0;
  36.             }
  37.         }).sum(1);
  38.  
  39.         // 打印输出结果
  40.         sum.print();
  41.  
  42.         // 执行任务
  43.         env.execute();
  44.     }
  45. }

        首先创建一个Flink的执行环境，并添加了自定义的数据源RandomFlowSource用作输入流。然后，通过MapFunction将输入的字符串转换成包含交易类型和金额的元组数据流，并按照交易类型进行分组，对交易金额进行求和。最后，打印输出结果并执行整个任务。

本地运行任务

        本地运行任务就是在IDEA中直接运行任务程序，此时在控制台中将源源不断地输出计算的结果：

        在这里我们之间将统计结果输出到了控制台中，如果想要输出到其他的地方就需要自定义Sink，或者通过Flink的其他输出依赖包将内容输出到其他的地方。

集群运行任务

        要想在集群中运行Flink任务，我们需要将我们前面写的Flink代码打包成jar包并通过Flink UI提交。

        提交之前将依赖的生命周期修改成provided：

  1. <properties>
  2.     <base.flink.scope>provided</base.flink.scope>
  3. </properties>

        之后使用Maven将代码打包：

        通过UI将jar包上传到集群上：

        双击jar包完成上传：

        上传完成后可以看到我们上传到集群中的jar包：

        之后点击需要运行的jar包，输入并行数、主程序等开始执行任务：

        可以看到我们的任务已经成功开始执行了，如果想要查看输出的结果可以在Flink TaskManager中的logs看到（一般来说输出的内容在stdout中查看，但是由于特性由Docker创建的Flink集群不能直接在stdout中看到）。

docker logs -f <你的taskmanager结点的容器名>

        在本文中的taskmanager结点的容器名分别是taskmanager-1和taskmanager-2，在控制台中输入命令可以看到输出的结果如下：

        至此，对Docker下搭建的Flink集群就介绍到这里了。

上一篇：Docker下的Storm

下一篇：Docker下的Zookeeper集群

点击获得全部Docker大数据专栏文章

本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号