Flink清洗日志服务SLS的数据并求ACUPCU

article/2025/8/29 11:51:22

上文说到为什么使用Flink实时消费阿里云日志服务SLS的数据，并把阿里云上Flink消费SLS的代码粘贴到本地，做了相关修改之后成功把整个流程跑通了。但仅仅这样是不够的，从控制台上面输出的数据来看是个比较难看的字符串，可以说没多大用处。因此本文主要是继续使用Flink来对从日志服务SLS过来的数据做一系列的清洗，然后再计算几个工作中的指标。

相关ETL代码如下，就是把需要使用到的各个字段提取出来，后期做迭代的时候再把下面代码重新封装一下

public static final class EtlFlatMapFunction implements FlatMapFunction<RawLogGroupList, Tuple3<String, String, Long>> {@Overridepublic void flatMap(RawLogGroupList rawLogGroupList, Collector<Tuple3<String, String, Long>> collector) throws Exception {Iterator rawLogGroupsIterator = rawLogGroupList.rawLogGroups.iterator();while (rawLogGroupsIterator.hasNext()) {RawLogGroup lg = (RawLogGroup) rawLogGroupsIterator.next();// String source = lg.source;// Map<String, String> tags = lg.tags;String topic = lg.topic;Iterator<RawLog> rowLogs = lg.logs.iterator();while (rowLogs.hasNext()) {RawLog rowLog = rowLogs.next();Map<String, String> contents = rowLog.getContents();Long online = Long.parseLong(contents.getOrDefault("online", "0"));String logTime = contents.getOrDefault("time", "0");String changeLogTime = logTime.substring(0, 17) + "00";collector.collect(new Tuple3<>(topic, changeLogTime, online));}}}}

从上图可以看到，ETL成功了。

下面我们一起来算ACU和PCU两个指标

代码写好之后，检查了一遍又一遍，代码的逻辑没有问题，但就是如下图那样标红，当时也比较郁闷，百思不得其解，因为我在另外一个编辑器使用AggregateFunction是完全没有问题的。

后来经过一番思考之后，恍然大悟，Flink的版本问题！日志服务SLS这个source支持的Flink版本是1.3.2，而我另外一个编辑器的Flink版本是1.7.2。想到之后，瞬间无语。

因为不同的Flink版本，AggregateFunction的返回值的类型是不一样的，如下图是Flink1.3.2中add的返回值类型

下图是Flink1.7.2中add的返回值类型

这时候，我想耍一个小心机，能不能直接把Flink的版本升级到1.7.2呢？答案是不行，如下图所示，少包了。

那没有办法，还是用回Flink1.3.2吧，然后把add的返回类型设置为void，并稍微修改add的逻辑

        @Overridepublic void add(Tuple3<String, String, Long> value, Tuple3<Long, Long, Long> accumulator) {
//            Long pcu = Math.max(value.f2, accumulator.f2);accumulator.f0 += value.f2;accumulator.f1 += 1L;accumulator.f2 = Math.max(value.f2, accumulator.f2);
//            return new Tuple3<>(accumulator.f0 + value.f2, accumulator.f1 + 1L, pcu);}

好，运行吧，成功！！

下面是整个程序的主体代码

package com.flink.onlinenumber;import com.aliyun.openservices.log.flink.FlinkLogConsumer;
import com.aliyun.openservices.log.flink.data.RawLogGroupList;
import com.aliyun.openservices.log.flink.data.RawLogGroupListDeserializer;
import com.aliyun.openservices.log.flink.util.Consts;
import com.flink.functions.OnlineNumberFunctions;
import com.flink.utils.SlsProps;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;import java.util.*;public class RealTimeOnlineNumApp {public static void main(String[] args) throws Exception {final Properties configProps = SlsProps.getSlsConfigProps("mafia-online_number", Consts.LOG_BEGIN_CURSOR);final OutputTag<Tuple3<String, String, Long>> lateOutputTag = new OutputTag<Tuple3<String, String, Long>>("acu-puc-late-data"){};// 设置日志服务的消息反序列化方法RawLogGroupListDeserializer deserializer = new RawLogGroupListDeserializer();final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 开启Flink exactly once语义env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);// 每5s保存一次checkpointenv.enableCheckpointing(5000);env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStream<RawLogGroupList> logStream = env.addSource(new FlinkLogConsumer<>(deserializer, configProps));DataStream<Tuple3<String, String, Long>> etlLogStream = logStream.flatMap(new OnlineNumberFunctions.EtlFlatMapFunction());// 入库MySQLetlLogStream.print();// 求acu pcuDataStream<Tuple3<String, String, Long>> etlLogStreamWithTsWm = etlLogStream.assignTimestampsAndWatermarks(new OnlineNumberFunctions.AcuPcuWatermarksPeriodicGenerator());etlLogStreamWithTsWm.keyBy(0).timeWindow(Time.minutes(30)).allowedLateness(Time.minutes(5))
//                .sideOutputLateData(lateOutputTag)
//                .aggregate(new OnlineNumberFunctions.AcuPcuAggregateFunc())
//                .reduce(new EtlLogReduceFunc())
//                .process(new OnlineNumberFunctions.EtlLogProcessWindowFunction()).aggregate(new OnlineNumberFunctions.AcuPcuAggregateFunc(),new OnlineNumberFunctions.EtlLogReduceProcessWindowFunction()).print();.getSideOutput(lateOutputTag)env.execute("RealTimeOnlineNumApp");}}

可以看到，上面的代码我已经做了一些优化，比如说，我一开始是使用的ProcessWindowFunction来做聚合的，但是ProcessWindowFunction是等到窗口触发的时候把窗口内所有的数据做聚合，显然，这样的性能是不高的，当然我可以直接使用reduce来直接做增量聚合，这样的话每来一条数据就处理一条，但是问题来了，这个需求还是需要返回窗口开始和结束的时间，而这是reduce并没有提供的，因此，最后选择了aggregate，既可以做增量聚合，又可以获取到窗口的相关元信息，只需要把AggregateFunction和ProcessWindowFunction的对象作为参数传进aggregate方法就可以了，注意，如果aggregate只有一个参数，它也仅仅是增量聚合而已。

下一篇，我会对之前写的代码进行迭代和优化，敬请期待！！