基于flink,kafka和vue的大数据股票看板

在向flink的网页中提交一个任务，点击Show Plan出现如下错误：
 进入linux查看日记信息：
cat flink-fei-standalonesession-0-hadoop102.log.1 
2022-03-27 01:00:56,287 WARN  akka.remote.ReliableDeliverySupervisor                       [] - Association with remote system [akka.tcp://flink@192.168.10.103:38496] has failed, address is now gated for [50] ms. Reason: [Association failed with [akka.tcp://flink@192.168.10.103:38496]] Caused by: [java.net.ConnectException: 拒绝毗连: /192.168.10.103:38496]
2022-03-27 01:00:57,803 WARN  akka.remote.ReliableDeliverySupervisor                       [] - Association with remote system [akka.tcp://flink@192.168.10.104:43792] has failed, address is now gated for [50] ms. Reason: [Disassociated] 
2022-03-27 01:00:57,804 WARN  akka.remote.ReliableDeliverySupervisor                       [] - Association with remote system [akka.tcp://flink-metrics@192.168.10.104:43949] has failed, address is now gated for [50] ms. Reason: [Disassociated] 
2022-03-27 01:00:58,754 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - RECEIVED SIGNAL 15: SIGTERM. Shutting down as requested.
2022-03-27 01:00:58,769 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint        [] - Shutting StandaloneSessionClusterEntrypoint down with application status UNKNOWN. Diagnostics Cluster entrypoint has been closed externally..
2022-03-27 01:00:58,778 INFO  org.apache.flink.runtime.dispatcher.DispatcherRestEndpoint   [] - Shutting down rest endpoint.
2022-03-27 01:00:58,780 INFO  org.apache.flink.runtime.blob.BlobServer                     [] - Stopped BLOB server at 0.0.0.0:45143
2022-03-27 01:00:59,096 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator        [] - Shutting down remote daemon.
2022-03-27 01:00:59,096 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator        [] - Remote daemon shut down; proceeding with flushing remote transports.
2022-03-27 01:00:59,097 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator        [] - Shutting down remote daemon.
2022-03-27 01:00:59,115 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator        [] - Remote daemon shut down; proceeding with flushing remote transports.
2022-03-27 01:00:59,144 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator        [] - Remoting shut down.
2022-03-27 01:00:59,152 INFO  akka.remote.RemoteActorRefProvider$RemotingTerminator        [] - Remoting shut down.
 通过观察是拒绝毗连，回到代码中看到原来着实监听一个端口输入的数据造成的错误
        //从参数中提取主机名和端标语
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String hostName = parameterTool.get("host");
        Integer port = parameterTool.getInt("port");
 
        //读取文本流
        DataStreamSource<String> lineDataStream = env.socketTextStream(hostName, port);
 由于还是实验阶段，并没有真正的通过监听端口获取数据来计算，这里采用给定一个地点和端标语的方式来实现。即如下方式来实现：
DataStreamSource<String> lineDataStream = env.socketTextStream("hadoop102", 7777);
重新编译并打包，将达成的jar包上传到flink的网页中。
将该填的信息填好，点击Show Plan弹出下面的页面：
下一步点击Submit提交任务执行，又出现错误：
通过查看异常信息知道：

 出现上面异常的原因是在linux上还没有执行如下命令：
nc -lk 7777
 在hadoop102的虚拟机上执行上面的一行命令，之后重新提交一次任务

这下可以大概正常运行了，之后在linux中输入一行数据，测试程序是否好用。 
linux输入一行数据用“ ”隔开

 程序正在运行：
 在TaskManager中查看执行结果
 从上面可以看到任务着实在个节点上运行的，这也是符合集群运行的特点的。
                        
原文链接：https://blog.csdn.net/weixin_46005650/article/details/123767403

1. * 交易平台排名服务类
2. * 负责统计和排名各个交易平台的交易量
3. * 使用Flink进行流式计算，并将结果存储到Redis中的有序集合(ZSET)
4. * 支持实时更新平台交易量排名
5. */
6. package com.example.service;
8. import com.example.Utils.Env;
9. import com.example.entity.StockTrade;
11. import jakarta.annotation.PostConstruct;
12. import jakarta.annotation.Resource;
13. import lombok.extern.slf4j.Slf4j;
14. import org.apache.flink.api.common.functions.AggregateFunction;
15. import org.apache.flink.api.common.functions.MapFunction;
16. import org.apache.flink.api.java.tuple.Tuple2;
17. import org.apache.flink.configuration.Configuration;
18. import org.apache.flink.streaming.api.datastream.DataStream;
19. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
20. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
21. import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
22. import org.apache.flink.streaming.api.windowing.time.Time;
23. import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
24. import org.springframework.stereotype.Component;
25. import redis.clients.jedis.Jedis;
27. @Component
28. @Slf4j
29. public class TopTenPlatform {
30.     /** 环境配置依赖注入 */
31.     @Resource
32.     private Env env_;
34.     /**
35.      * 初始化方法，在服务启动时自动执行
36.      * 在新线程中启动Flink作业
37.      */
38.     @PostConstruct
39.     public void init() {
40.         new Thread(() -> {
41.             try {
42.                 startFlinkJob();
43.             } catch (Exception e) {
44.                 log.error("Error starting Flink job", e);
45.             }
46.         }).start();
47.     }
49.     /**
50.      * 启动Flink作业
51.      * 设置数据流处理管道，包括数据转换、窗口操作和数据聚合
52.      * @throws Exception 如果作业启动失败
53.      */
55.     /*
56.     * 这段代码是一个Apache Flink的流处理作业，它的作用是分析股票交易数据，并统计每秒钟每个交易平台的交易量。以下是代码的详细解释：
58. 1. **获取执行环境**：
59.    ```java
60.    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
61.    ```
62.    这行代码获取了Flink的`StreamExecutionEnvironment`，它是执行Flink作业的入口点。
64. 2. **获取数据流**：
65.    ```java
66.    DataStream<StockTrade> tradeStream = env_.getEnv(env);
67.    ```
68.    这行代码从`env_`对象（应该是一个你自己定义的类，包含Flink环境配置）获取一个`DataStream`，这个流包含了`StockTrade`类型的数据。
70. 3. **记录日志**：
71.    ```java
72.    log.info("数据源接入");
73.    ```
74.    这行代码记录了一条日志信息，表明数据源已经接入。
76. 4. **转换数据流**：
77.    ```java
78.    tradeStream
79.        .map(new MapFunction<StockTrade, Tuple2<String, Long>>() {
80.            @Override
81.            public Tuple2<String, Long> map(StockTrade value) throws Exception {
82.                return new Tuple2<>(value.getTradePlatform(), value.getTradeVolume());
83.            }
84.        })
85.    ```
86.    这部分代码将`DataStream<StockTrade>`转换为`DataStream<Tuple2<String, Long>>`。`map`函数将每个`StockTrade`对象转换为一个包含平台名称和交易量的二元组（`Tuple2`）。
88. 5. **按键分组**：
89.    ```java
90.    .keyBy(value -> value.f0)
91.    ```
92.    这行代码按二元组的第一个元素（平台名称）对数据流进行分组，为接下来的窗口操作准备。
94. 6. **开窗操作**：
95.    ```java
96.    .window(TumblingEventTimeWindows.of(Time.seconds(1)))
97.    ```
98.    这行代码应用了一个基于事件时间的滚动窗口，窗口大小为1秒。这意味着每1秒钟的数据会被分到一个窗口中进行处理。
100. 7. **聚合操作**：
101.    ```java
102.    .aggregate(new TradeAggregator())
103.    ```
104.    这行代码应用了一个自定义的聚合函数`TradeAggregator`，它将对每个窗口内的数据进行聚合操作，比如累加每个交易平台的交易量。
106. 8. **添加Sink**：
107.    ```java
108.    .addSink(new AccumulatingRedisSink(redisConfig));
109.    ```
110.    这行代码将聚合结果输出到Redis数据库。`AccumulatingRedisSink`是一个自定义的Sink，它负责将聚合结果持久化到Redis。
112. 9. **执行作业**：
113.    ```java
114.    System.out.println("Top 10 stock trade volume analysis started");
115.    env.execute("Stock Trade Analysis");
116.    ```
117.    这两行代码启动了Flink作业。`env.execute`方法会根据提供的作业名称（"Stock Trade Analysis"）执行定义好的流处理作业。
119. 总结来说，这段代码定义了一个Flink流处理作业，它从某个数据源获取股票交易数据，将数据转换为平台名称和交易量的二元组，然后按平台名称分组，对每组数据应用1秒的滚动窗口，聚合每个窗口内的交易量，并将结果输出到Redis数据库。这个作业可以帮助分析每个交易平台每秒钟的交易量。
121.     * */
123.     public void startFlinkJob() throws Exception {
124.         // 获取Flink的流执行环境，这是执行Flink作业的配置环境
125.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
127.         // 从env_对象获取DataStream，该对象是StockTrade类型的数据流
128.         DataStream<StockTrade> tradeStream = env_.getEnv(env);
130.         // 记录日志信息，表明数据源已经接入
131.         log.info("数据源接入");
133.         // 对DataStream进行处理，1小时窗口的操作
134.         tradeStream
135.                 // 将StockTrade对象转换为平台名称和交易量的二元组
136.                 .map(new MapFunction<StockTrade, Tuple2<String, Long>>() {
137.                     /**
138.                      * 将StockTrade对象转换为平台名称和交易量的二元组
139.                      * @param value 股票交易数据
140.                      * @return 包含平台名称和交易量的二元组
141.                      */
142.                     @Override
143.                     public Tuple2<String, Long> map(StockTrade value) throws Exception {
144.                         // 返回一个包含平台名称和交易量的二元组
145.                         return new Tuple2<>(value.getTradePlatform(), value.getTradeVolume());
146.                     }
147.                 })
148.                 // 按照平台名称（二元组的第一个元素）对数据进行分组
149.                 .keyBy(value -> value.f0)
150.                 // 应用1秒的滚动事件时间窗口
151.                 .window(TumblingEventTimeWindows.of(Time.seconds(1)))
152.                 // 对每个窗口内的数据进行聚合操作，使用自定义的TradeAggregator聚合函数
153.                 .aggregate(new TTPLTradeAggregator())
154.                 // 将聚合结果输出到Redis数据库，使用自定义的AccumulatingRedisSink作为Sink
155.                 .addSink(new TTPLAccumulatingRedisSink(redisConfig));
157.         // 在控制台打印日志，表明作业开始执行
158.         System.out.println("Top 10 stock trade volume analysis started");
160.         // 执行Flink作业，作业名为"Stock Trade Analysis"
161.         env.execute("Stock Trade Analysis");
162.     }
164.     /** Redis连接配置 */
165.     public FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
166.             .setHost("localhost")
167.             .setPort(6379)
168.             .build();
170.     /**
171.      * Redis数据写入Sink
172.      * 负责将平台交易量数据持久化到Redis的有序集合中
173.      * 使用ZSET数据结构实现自动排序
174.      */
175.     public static class TTPLAccumulatingRedisSink extends RichSinkFunction<Tuple2<String, Long>> {
176.         /** Redis客户端实例 */
177.         private transient Jedis jedis;
178.         /** Redis连接配置 */
179.         private final FlinkJedisPoolConfig config;
181.         /**
182.          * 构造函数
183.          * @param config Redis连接配置
184.          */
185.         public TTPLAccumulatingRedisSink(FlinkJedisPoolConfig config) {
186.             this.config = config;
187.         }
189.         /**
190.          * 初始化Redis连接
191.          * @param parameters 配置参数
192.          */
193.         @Override
194.         public void open(Configuration parameters) throws Exception {
195.             super.open(parameters);
196.             jedis = new Jedis(config.getHost(), config.getPort());
197.         }
199.         /**
200.          * 关闭Redis连接
201.          */
202.         @Override
203.         public void close() throws Exception {
204.             if (jedis != null) {
205.                 jedis.close();
206.             }
207.             super.close();
208.         }
210.         /**
211.          * 处理每条平台交易量数据
212.          * 将数据累加到Redis的有序集合中
213.          * @param value 平台交易量数据
214.          * @param context 上下文
215.          */
216.         @Override
217.         public void invoke(Tuple2<String, Long> value, Context context) throws Exception {
218.             String hashKey = "stock_trade_platform"; // 存储股票交易量的键
219.             String field = value.f0; // 使用股票平台作为字段
221.             // 检查键的类型
222.             String type = jedis.type(hashKey);
223.             if (!"zset".equals(type)) {
224.                 // 如果键不存在或类型不正确，删除旧的键
225.                 jedis.del(hashKey);
226.                 // 创建新的有序集合
227.                 jedis.zadd(hashKey, 0, field); // 初始化一个空的有序集合
228.             }
230.             // 从 Redis 中获取现有值
231.             Double existingVolume = jedis.zscore(hashKey, field);
232.             if (existingVolume == null) {
233.                 existingVolume = (double) 0L;
234.             }
236.             // 更新交易量
237.             double updatedVolume = existingVolume + value.f1;
239.             // 将更新后的值写回 Redis
240.             jedis.zadd(hashKey, updatedVolume, field);
241.             final long delayInterval = 100;
242.             Thread.sleep(delayInterval);
243.         }
244.     }
246.     /**
247.      * 交易量聚合器
248.      * 实现AggregateFunction接口，用于聚合计算平台交易量
249.      */
250.     public static class TTPLTradeAggregator implements AggregateFunction<Tuple2<String, Long>, Tuple2<String, Long>, Tuple2<String, Long>> {
251.         /**
252.          * 创建新的累加器
253.          * @return 空的累加器
254.          */
255.         @Override
256.         public Tuple2<String, Long> createAccumulator() {
257.             return new Tuple2<>("", 0L);
258.         }
260.         /**
261.          * 将新的交易数据添加到累加器中
262.          * @param value 新的交易数据
263.          * @param accumulator 当前累加器
264.          * @return 更新后的累加器
265.          */
266.         @Override
267.         public Tuple2<String, Long> add(Tuple2<String, Long> value, Tuple2<String, Long> accumulator) {
268.             String stockCode = value.f0;
269.             Long totalAmount = value.f1 + accumulator.f1;
270.             return new Tuple2<>(stockCode, totalAmount);
271.         }
273.         /**
274.          * 获取聚合结果
275.          * @param accumulator 累加器
276.          * @return 聚合结果
277.          */
278.         @Override
279.         public Tuple2<String, Long> getResult(Tuple2<String, Long> accumulator) {
280.             return accumulator;
281.         }
283.         /**
284.          * 合并两个累加器
285.          * @param a 第一个累加器
286.          * @param b 第二个累加器
287.          * @return 合并后的累加器
288.          */
289.         @Override
290.         public Tuple2<String, Long> merge(Tuple2<String, Long> a, Tuple2<String, Long> b) {
291.             String stockCode = a.f0; // 假设两个 accumulator 的股票代码相同
292.             Long totalAmount = a.f1 + b.f1;
293.             return new Tuple2<>(stockCode, totalAmount);
294.         }
295.     }
296. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。