这次来整个高端的API实时QPS流计算

package org.myorg.quickstart;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class BatchJob {

public static void main(String[] args) throws Exception {
// set up the batch execution environment
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<String> text = env.readTextFile("/path/xxxx.log");

        DataSet<Tuple2<String, Integer>> counts =
                text.flatMap(new LogLevelFilter())
                        .groupBy(0)
                        .sum(1);

        counts.print();

        env.execute("Flink Batch Java API Skeleton");
    }

// 自定义函数
public static class LogLevelFilter implements FlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
// json decode每一行的log
            JSONObject jsonObject = JSON.parseObject(value);
// 统计不同level的log
if (jsonObject.containsKey("level")) {
String level = jsonObject.getString("level");
                out.collect(new Tuple2<>(level, 1));
            }
        }
    }
}

{
"@timestamp":"2020-12-27T18:58:38+08:00",
"remote_addr":"127.0.0.1",
"referer":"-",
"request":"GET /wechat/config HTTP/1.1",
"status":200,
"bytes":64,
"user_agent":"Mozilla/5.0 (Macintosh; Intel Mac OS X 11_0_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.88 Safari/537.36",
"http_x_forwarded":"-",
"request_time":"0.145"
}

package org.myorg.quickstart;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class StreamingJob {

public static final Logger logger = LoggerFactory.getLogger(StreamingJob.class);

public static void main(String[] args) throws Exception {
// set up the streaming execution environment
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

TextInputFormat textInputFormat = new TextInputFormat(null);

DataStream<String> nginxAccessLog = env.readFile(textInputFormat, "path/to/nginx_log/access.log", FileProcessingMode.PROCESS_CONTINUOUSLY, 100);

DataStream<StatusCount> nginxAccessLogStream = nginxAccessLog.flatMap((FlatMapFunction<String, StatusCount>) (s, collector) -> {
try {
JSONObject jsonObject = JSON.parseObject(s);
if (jsonObject.containsKey("status")) {
Integer status = jsonObject.getInteger("status");
if (status.compareTo(500) < 0) {
                        collector.collect(StatusCount.of(status + "", 1L));
                    }
                }
            } catch (Exception e) {
                logger.error("解析nginx access log 错误 {}", e.getMessage());
            }
        });
DataStream<StatusCount> windowsCounts = nginxAccessLogStream.
                keyBy("status").
                timeWindow(Time.seconds(1)).
                sum("count");
        windowsCounts.print().setParallelism(1);

        env.execute("Flink Streaming Java API Skeleton");
    }

public static class StatusCount {

public String status;
public Long count;

public static StatusCount of(String word, Long count) {
StatusCount statusCount = new StatusCount();
            statusCount.status = word;
            statusCount.count = count;
return statusCount;
        }

        @Override
public String toString() {
return "status:" + status + " count:" + count;
        }
    }
}