死磕flink(四)

// 输入路径和输出路径通过程序的参数传入  // 第一个参数是输入路径（文本文件的路径），第二个参数是输出路径（结果保存的路径）  String inPath = args[0];  String outPath = args[1];  // 获取 Flink 的批处理执行环境  // 批处理执行环境用于管理 Flink 的数据流和算子  ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();  // 从指定的输入路径读取文本文件中的数据  // readTextFile 方法返回一个 DataSet，其中每一条记录是文本文件中的一行 DataSet<String> text = executionEnvironment.readTextFile(inPath);

// 对读取的数据进行处理// 1. 使用 FlatMapFunction 对每一行进行处理，将每行拆分成单词，并标记每个单词出现的次数（1）// 2. 使用 groupBy 按照单词进行分组// 3. 使用 sum 聚合每个单词的出现次数// 结果是一个 DataSet，其中包含每个单词及其出现的总次数DataSet<Tuple2<String, Integer>> dataSet = text.flatMap(new LineSplitter()).groupBy(0).sum(1);

// 自定义 FlatMapFunction 实现类，用于将每行文本拆分为单词，并创建 (单词, 1) 的 Tuple2 对象static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {       @Override   public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {      // 按空格拆分每行文本，得到每个单词      for (String word : line.split(" ")) {          // 将每个单词与计数 1 组成一个 Tuple2 对象，并收集        collector.collect(new Tuple2<String, Integer>(word, 1));       }    }  }

// 将处理结果写入到指定的输出路径// writeAsCsv 方法将结果以 CSV 格式写入到文件中// \n 表示行分隔符，"" 表示字段分隔符// setParallelism(1) 设置并行度为 1，确保所有结果写入到一个文件中 dataSet.writeAsCsv(outPath, "\n", "").setParallelism(1);// 触发作业的执行// execute 方法启动 Flink 作业，并使用指定的名称executionEnvironment.execute("wordcount batch process");

import java.io.IOException;import java.io.OutputStream;import java.net.ServerSocket;import java.net.Socket;import java.text.SimpleDateFormat;import java.util.Date;import java.util.Scanner;import java.util.concurrent.TimeUnit;
public class NcServer {
    // 定义端口号    private static final int PORT = 9000;
    public static void main(String[] args) throws IOException {
        ServerSocket server = new ServerSocket(PORT);        Socket socket = server.accept();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");        System.out.println("[" + simpleDateFormat.format(new Date()) + "]" + socket.getInetAddress() + "已建立连接！");
        //监控连接是否断开线程        new Thread(new CheckClientThread(socket)).start();
        //输出流        OutputStream outputStream = socket.getOutputStream();
        //控制台输入        Scanner in = new Scanner(System.in);
        while (true) {            String str = in.nextLine() + "\n";
            outputStream.write(str.getBytes());            outputStream.flush();        }
    }
    // 监控连接程序是否断开 线程类    static class CheckClientThread implements Runnable {
        private Socket socketClient;
        public CheckClientThread(Socket socketClient) {            this.socketClient = socketClient;        }
        public void run() {            while (true) {                try {                    TimeUnit.SECONDS.sleep(1);//                    socketClient.sendUrgentData(0xFF);                } catch (Exception e) {                    e.printStackTrace();                    SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");                    System.out.println("[" + simpleDateFormat.format(new Date()) + "]" + socketClient.getInetAddress() + "连接已关闭！");                    // 断开后退出程序                    System.exit(0);                }            }        }    }}

import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.java.tuple.Tuple2;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.streaming.api.windowing.time.Time;import org.apache.flink.util.Collector;
/** *   Socket模拟实时发送单词，使用Flink实时接收数据， *   对指定时间窗口内（如5s）的数据进行聚合统计，每隔1s汇总计算一次，并且把时间窗口内计算结果打印出来。 */public class WordCountStreamWindow {
    public static void main(String[] args) throws Exception {        // 监听的 IP 和端口号，默认值为 localhost 和 9000        // 可以通过命令行参数传入 IP 和端口号        String ip = "127.0.0.1";        int port = 9000;
        // 获取 Flink 的流执行环境        // StreamExecutionEnvironment 是 Flink 流处理作业的核心环境        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        // 从指定的 IP 和端口创建一个数据流源        // socketTextStream 方法返回一个 DataStream，其中每条记录是通过 socket 接收到的一行文本        DataStreamSource<String> textStream = streamExecutionEnvironment.socketTextStream(ip, port, "\n");
        // 使用 FlatMapFunction 对每行文本进行处理        // 将每行文本拆分成单词，并为每个单词生成一个 (单词, 1) 的 Tuple2 对象        SingleOutputStreamOperator<Tuple2<String, Long>> tuple2SingleOutputStreamOperator = textStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {            @Override            public void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {                // 按空格拆分每行文本，得到每个单词                String[] splits = s.split("\\s");                // 遍历每个单词，将每个单词与计数 1 组成一个 Tuple2 对象，并收集                for (String word : splits) {                    collector.collect(Tuple2.of(word, 1L));                }            }        });
        // 按单词进行分组，并对每个单词的计数进行聚合        // keyBy(0) 按照 Tuple2 的第 0 个元素（即单词）进行分组        // timeWindow(Time.seconds(2), Time.seconds(1)) 使用 2 秒的时间窗口，每秒钟滑动一次窗口        // sum(1) 对每个分组中的第 1 个元素（即计数）进行求和        SingleOutputStreamOperator<Tuple2<String, Long>> word = tuple2SingleOutputStreamOperator.keyBy(0)                .timeWindow(Time.seconds(2), Time.seconds(1))  // 使用 2 秒的时间窗口，每秒钟滑动一次窗口                .sum(1);
        // 打印计算结果到标准输出        // 输出每个时间窗口的单词计数结果        word.print();
        // 启动 Flink 作业，提交到 Flink 执行环境进行处理        streamExecutionEnvironment.execute("wordcount stream process");    }}