2、Flink DataStreamAPI 概述（下）

代码示例

Maven 依赖

        
            org.apache.flink
            flink-streaming-java
            1.19.0
        
        
            org.apache.flink
            flink-clients
            1.19.0
        
        
            org.slf4j
            slf4j-api
            1.7.25
        
        
            org.slf4j
            slf4j-log4j12
            1.7.25
        
    

log4j.properties

log4j.rootLogger=INFO, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n

1、_01_QuickStart

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.util.Collector;
import java.time.Duration;
public class _01_QuickStart {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream dataStream = env
                .socketTextStream("localhost", 8888)
                .flatMap(new Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5)))
                .sum(1);
        dataStream.print();
//        JobExecutionResult jobExecutionResult = env.execute("Window WordCount");
        //程序完成时打印
        //JobExecutionResult=>Program execution finished
        //Job with JobID c01e1255752cbb34469a9a10177e637c has finished.
        //Job Runtime: 25596 ms
//        System.out.println("JobExecutionResult=>"+jobExecutionResult.getJobExecutionResult());
        JobClient jobClient = env.executeAsync("Window WordCount");
        // Java程序可以通过JobClient同Flink Job交互
        // jobID=>32e976f03ac7243c09a5cf07c0739921
        // jobStatus=>RUNNING
        System.out.println("jobID=>"+jobClient.getJobID());
        System.out.println("jobStatus=>"+jobClient.getJobStatus().get());
    }
    public static class Splitter implements FlatMapFunction {
        @Override
        public void flatMap(String sentence, Collector out) throws Exception {
            for (String word : sentence.split(",")) {
                out.collect(new Tuple2(word, 1));
            }
        }
    }
}

2、_02_ReadFileSource

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import java.io.File;
public class _02_ReadFileSource {
    public static void main(String[] args) throws Exception {
        ExecutionConfig executionConfig = new ExecutionConfig();
        executionConfig.setAutoWatermarkInterval(1000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(executionConfig.toConfiguration());
        env.setBufferTimeout(1000);
        System.out.println("自动生成水位线间隔=>"+env.getConfig().getAutoWatermarkInterval());
        //第一次打印
        //1> a
        //5> h
        //6> i
        //4> f
        //2> c
        //3> e
        //1> b
        //2> d
        //4> g
        //向本地文件中新增三行insert
        //第二次打印
        //5> insert
        //2> i
        //2> insert
        //3> insert
        //1> g
        //1> h
        //8> d
        //7> a
        //8> e
        //7> b
        //8> f
        //7> c
        DataStreamSource source = env.readFile(new TextInputFormat(Path.fromLocalFile(new File("word.txt")))
                , "/Users/***/Desktop/word.txt"
                , FileProcessingMode.PROCESS_CONTINUOUSLY
                , 10000);
        source.print();
        env.execute();
    }
}

3、_03_CollectAsync

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.CloseableIterator;
public class _03_CollectAsync {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 从元素列表创建一个 DataStream
        DataStream myInts = env.fromElements(1, 2, 3, 4, 5);
        CloseableIterator iterator = myInts.collectAsync();
        env.execute();
        while (iterator.hasNext()){
            System.out.println("iterator=>"+iterator.next());
        }
    }
}

4、_04_JobClientStopWithSavepoint

import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.util.CloseableIterator;
import java.io.File;
// 在 程序完成时 或者 CheckPoint触发时 才会输出结果
public class _04_JobClientStopWithSavepoint {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(2000);
        DataStreamSource source = env.readFile(new TextInputFormat(Path.fromLocalFile(new File("word.txt")))
                , "/Users/***/Desktop/word.txt"
                , FileProcessingMode.PROCESS_CONTINUOUSLY
                , 10000);
        CloseableIterator iterator = source.collectAsync();
        JobClient jobClient = env.executeAsync();
//        TimeUnit.SECONDS.sleep(5);
//        jobClient.stopWithSavepoint(false,"/Users/hhx/Desktop/", SavepointFormatType.DEFAULT);
        while (iterator.hasNext()){
            System.out.println("iterator=>"+iterator.next());
        }
    }
}