代码示例

Maven 依赖

<dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>1.19.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>1.19.0</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.25</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.25</version></dependency></dependencies>

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<Tuple2<String, Integer>> 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=>RUNNINGSystem.out.println("jobID=>"+jobClient.getJobID());System.out.println("jobStatus=>"+jobClient.getJobStatus().get());}public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {@Overridepublic void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {for (String word : sentence.split(",")) {out.collect(new Tuple2<String, Integer>(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> cDataStreamSource<String> 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();// 从元素列表创建一个 DataStreamDataStream<Integer> myInts = env.fromElements(1, 2, 3, 4, 5);CloseableIterator<Integer> 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<String> source = env.readFile(new TextInputFormat(Path.fromLocalFile(new File("word.txt"))), "/Users/***/Desktop/word.txt", FileProcessingMode.PROCESS_CONTINUOUSLY, 10000);CloseableIterator<String> 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());}}
}