Flink 的 DataSet 和 DataStream 的 API，并模拟了实时计算的场景。

（一）说好的流批一体呢

现状

Flink 很重要的一个特点是“流批一体”，然而事实上 Flink 并没有完全做到所谓的“流批一体”，即编写一套代码，可以同时支持流式计算场景和批量计算的场景。目前截止 1.10 版本依然采用了 DataSet 和 DataStream 两套 API 来适配不同的应用场景。

DateSet 和 DataStream 的区别和联系

在官网或者其他网站上，都可以找到目前 Flink 支持两套 API 和一些应用场景，但大都缺少了“为什么”这样的思考。

Apache Flink 在诞生之初的设计哲学是：用同一个引擎支持多种形式的计算，包括批处理、流处理和机器学习等。尤其是在流式计算方面，Flink 实现了计算引擎级别的流批一体。那么对于普通开发者而言，如果使用原生的 Flink ，直接的感受还是要编写两套代码。

整体架构如下图所示：

在 Flink 的源代码中，我们可以在 flink-java 这个模块中找到所有关于 DataSet 的核心类，DataStream 的核心实现类则在 flink-streaming-java 这个模块。

在上述两张图中，我们分别打开 DataSet 和 DataStream 这两个类，可以发现，二者支持的 API 都非常丰富且十分类似，比如常用的 map、filter、join 等常见的 transformation 函数。

对于 DataSet 而言，Source 部分来源于文件、表或者 Java 集合；而 DataStream 的 Source 部分则一般是消息中间件比如 Kafka 等。

（二）Environment

Flink有以下几种Environment

1. 批处理Environment，ExecutionEnvironment

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

2.流处理Environment，StreamExecutionEnvironment

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

3. 本机Environment，LocalEnvironment

ExecutionEnvironment env = LocalEnvironment.getExecutionEnvironment();

4. java集合Environment，CollectionEnvironment

ExecutionEnvironment env = CollectionEnvironment.getExecutionEnvironment();

创建Environment的方法

1. getExecutionEnvironment ，含义就是本地运行就是 createLocalEnvironment，如果是通过client提交到集群上，就返回集群的环境

Creates an execution environment that represents the context in which the program is currently executed.
    * If the program is invoked standalone, this method returns a local execution environment, as returned by
    * {@link #createLocalEnvironment()}. If the program is invoked from within the command line client to be
    * submitted to a cluster, this method returns the execution environment of this cluster.

2. createLocalEnvironment ，返回本地执行环境，需要在调用时指定默认的并行度，比如

LocalStreamEnvironment env1 = StreamExecutionEnvironment.createLocalEnvironment(1);
 
LocalEnvironment env2 = ExecutionEnvironment.createLocalEnvironment(1);

3. createRemoteEnvironment， 返回集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号，并指定要在集群中运行的 Jar 包，比如

StreamExecutionEnvironment env1 = StreamExecutionEnvironment.createRemoteEnvironment("127.0.0.1", 8080, "/path/word_count.jar");
 
ExecutionEnvironment env2 = ExecutionEnvironment.createRemoteEnvironment("127.0.0.1", 8080, "/path/word_count.jar");

（三）DataStream API

DataStream是Flink编写流处理作业的API。我们前面说过一个完整的Flink处理程序应该包含三部分：数据源(Source)、转换操作(Transformation)、结果接收(Sink)。下面我们从这三部分来看DataStream API。

（四）数据源(Source)

Flink应用程序从数据源获取要处理的数据，DataStream通过StreamExecutionEnvironment.addResource(SourceFunction) 来添加数据源。为了方便使用，Flink预提几类预定义的数据源，比如读取文件的Source、通过Sockt读取的Source、从内存中获取的Source等。

1.基于集合的预定义Source

基于集合的数据源一般是指从内存集合中直接读取要处理的数据，StreamExecutionEnvironment提供了4类预定义方法。

fromCollection

fromCollection是从给定的集合中创建DataStream，StreamExecutionEnvironment提供了4种重载方法：

• fromCollection(Collection<T> data)：通过给定的集合创建DataStream。返回数据类型为集合元素类型。
• fromCollection(Collection<T> data,TypeInformation<T> typeInfo)：通过给定的非空集合创建DataStream。返回数据类型为typeInfo。
• fromCollection(Iterator<T> data,Class<T> type)：通过给定的迭代器创建DataStream。返回数据类型为type。
• fromCollection(Iterator<T> data,TypeInformation<T> typeInfo)：通过给定的迭代器创建DataStream。返回数据类型为typeInfo。

fromParallelCollection

fromParallelCollection和fromCollection类似，但是是并行的从迭代器中创建DataStream。

• fromParallelCollection(SplittableIterator<T> data,Class<T> type)
• fromParallelCollection(SplittableIterator<T>,TypeInfomation typeInfo)

和Iterable中Spliterator类似，这是JDK1.8新增的特性，并行读取集合元素。

fromElements

fromElements从给定的对象序列中创建DataStream，StreamExecutionEnvironment提供了2种重载方法：

• fromElements(T... data)：从给定对象序列中创建DataStream，返回的数据类型为该对象类型自身。
• fromElements(Class<T> type,T... data)：从给定对象序列中创建DataStream，返回的数据类型type。

generateSequence

generateSequence(long from,long to)从给定间隔的数字序列中创建DataStream，比如from为1，to为10，则会生成1~10的序列。

2.基于Socket的预定义Source

我们还可以通过Socket来读取数据，通过Sockt创建的DataStream能够从Socket中无限接收字符串，字符编码采用系统默认字符集。当Socket关闭时，Source停止读取。Socket提供了5个重载方法，但是有两个方法已经标记废弃。

• socketTextStream(String hostname,int port)：指定Socket主机和端口，默认数据分隔符为换行符(\n)。
• socketTextStream(String hostname,int port,String delimiter)：指定Socket主机和端口，数据分隔符为delimiter。
• socketTextStream(String hostname,int port,String delimiter,long maxRetry)：该重载方法能够当与Socket断开时进行重连，重连次数由maxRetry决定，时间间隔为1秒。如果为0则表示立即终止不重连，如果为负数则表示一直重试。

DataStreamSource<String> socketDS = env.socketTextStream("localhost", 7777);

3.基于文件的预定义Source

基于文件创建DataStream主要有两种方式：readTextFile和readFile。(readFileStream已废弃)。readTextFile就是简单读取文件，而readFile的使用方式比较灵活。

readTextFile

readTextFile提供了两个重载方法：

• readTextFile(String filePath)：逐行读取指定文件来创建DataStream，使用系统默认字符编码读取。
• readTextFile(String filePath,String charsetName)：逐行读取文件来创建DataStream，使用charsetName编码读取。

DataSource<String> lineDS = env.readTextFile("D:\\test\\test.txt");
DataStreamSource<String> lineDS = env.readTextFile("D:\\test\\test.txt");

readFile

readFile通过指定的FileInputFormat来读取用户指定路径的文件。对于指定路径文件，我们可以使用不同的处理模式来处理，FileProcessingMode.PROCESS_ONCE模式只会处理文件数据一次，而FileProcessingMode.PROCESS_CONTINUOUSLY会监控数据源文件是否有新数据，如果有新数据则会继续处理。

readFile(FileInputFormat<T> inputFormat,
　　　　　　String filePath,
　　　　　　　　FileProcessingMode watchType,
　　　　　　　　　　long interval,TypeInformation typrInfo) 

readFile提供了几个便于使用的重载方法，但它们最终都是调用上面这个方法的。

• readFile(FileInputFormat<T> inputFormat,String filePath)：处理方式默认使用FileProcessingMode.PROCESS_ONCE。
• readFile(FileInputFormat<T> inputFormat,String filePath,FileProcessingMode watchType,long interval)：返回类型默认为inputFormat类型。

需要注意：在使用FileProcessingMode.PROCESS_CONTINUOUSLY时，当修改读取文件时，Flink会将文件整体内容重新处理，也就是打破了"exactly-once"。

4.自定义Source

除了预定义的Source外，我们还可以通过实现SourceFunction来自定义Source，然后通过StreamExecutionEnvironment.addSource(sourceFunction)添加进来。比如读取Kafka数据的Source：

addSource(new FlinkKafkaConsumer08<>);

我们可以实现以下三个接口来自定义Source：

• SourceFunction：创建非并行数据源。
• ParallelSourceFunction：创建并行数据源。
• RichParallelSourceFunction：创建并行数据源。

（五）数据转换(Transformation)

数据处理的核心就是对数据进行各种转化操作，在Flink上就是通过转换将一个或多个DataStream转换成新的DataStream。

为了更好的理解transformation函数，下面给出匿名类的方式来实现各个函数。

所有转换函数都是依赖以下基础：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> dataStream = env.readTextFile("/opt/yjz/flink/flink-1.7.2/README.txt");

Aggregations

1.介绍

Aggregations 为聚合函数的总称，常见的聚合函数包括但不限于 sum、max、min 等。

事实上，对于 Aggregations 函数，Flink 帮助我们封装了状态数据，这些状态数据不会被清理，所以在实际生产环境中应该尽量避免在一个无限流上使用 Aggregations。而且，对于同一个 keyedStream ，只能调用一次 Aggregation 函数。

不建议的是那些状态无限增长的聚合，实际应用中一般会配合窗口使用。使得状态不会无限制扩张。

2.例子

滚动聚合具有相同key的数据流元素，我们可以指定需要聚合的字段(field)。DataStream<T>中的T为聚合之后的结果。

//对KeyedStream中元素的第一个Filed求和
DataStream<String> dataStream1 = keyedStream.sum(0);
//对KeyedStream中元素的“count”字段求和
keyedStream.sum("count");
//获取keyedStream中第一个字段的最小值
keyedStream.min(0);
//获取keyedStream中count字段的最小值的元素
keyedStream.minBy("count");
keyedStream.max("count");
keyedStream.maxBy(0);

min和minBy的区别是：min返回指定字段的最小值，而minBy返回最小值所在的元素。

转换前后数据类型：KeyedStream->DataStream。

broadcast

使用broadcast可以向每个分区广播元素。

dataStream.broadcast();

转换前后的数据类型：DataStream->DataStream。

CoFlatMap/CoMap

可以对连接流执行类似map和flatMap操作。

connectedStreams.map(new CoMapFunction<String, String, String>() {
            @Override
            public String map1(String value) throws Exception {
                return value.toUpperCase();
            }
            @Override
            public String map2(String value) throws Exception {
                return value.toLowerCase();
            }
        });

转换前后的数据类型：ConnectedDataStreams->DataStream。

Connect

连接(connect)两个流，并且保留其类型。两个数据流之间可以共享状态。ConnectedStreams<IN1,IN2>中IN1代表第一个数据流中的数据类型，IN2代表第二个数据流中的数据类型。

ConnectedStreams<String,String> connectedStreams = dataStream.connect(dataStream);

转换前后的数据类型：DataStream,DataStream->ConnectedDataStreams。

Extract Timestamps(Deprecated)

从记录中提取时间戳，以便使用事件时间语义窗口。之后会专门来看Flink的Event Time。

dataStream.assignTimestamps(new TimestampExtractor<String>() {
            @Override
            public long extractTimestamp(String element, long currentTimestamp) {
                return 0;
            }
            @Override
            public long extractWatermark(String element, long currentTimestamp) {
                return 0;
            }
            @Override
            public long getCurrentWatermark() {
                return 0;
            }
        });

该方法以被标记为废弃！

转换前后的数据类型：DataStream->DataStream。

Filter

1.介绍

顾名思义，Fliter 的意思就是过滤掉不需要的数据，每个元素都会被 filter 函数处理，如果 filter 函数返回 true 则保留，否则丢弃。

2.例子

过滤指定元素数据，如果返回true则该元素继续向下传递，如果为false则将该元素过滤掉。FilterFunction<T>中T代表输入元素的数据类型。

dataStream.filter(new FilterFunction<String>() {
    @Override
    public boolean filter(String line) throws Exception {
        if(line.contains("flink"))
            return true;
        else
            return false;
    }
});

转换前后数据类型：DataStream->DataStream。

FlatMap

1.介绍

FlatMap 接受一个元素，返回零到多个元素。接受一个元素作为输入，并且根据开发者自定义的逻辑处理后输出，但是当返回值是列表的时候，FlatMap 会将列表“平铺”，也就是以单个元素的形式进行输出。

2.例子

输入一个元素，输出0个、1个或多个元素。FlatMapFunction<T,V>中T代表输入元素数据类型(flatMap方法的第一个参数类型)，V代表输出集合中元素类型(flatMap中的Collector类型参数)

        FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Long>> out) throws Exception {
                String[] words = line.split(" ");

                for (String word : words) {
                    out.collect(Tuple2.of(word,1L));
                }
            }
        });

转换前后数据类型：DataStream->DataStream。

Fold(Deprecated)

Fold功能和Reduce类似，但是Fold提供了初始值，从初始值开始滚动对相同的key记录进行滚动合并。FoldFunction<T,V>中的T为KeyStream中元素数据类型，V为初始值类型和fold方法返回值类型。

keyedStream.fold(0, new FoldFunction<String, Integer>() {
    @Override
    public Integer fold(Integer accumulator, String value) throws Exception {
        return accumulator + value.split(" ").length;
    }
});

该方法已经标记为废弃！

转换前后数据类型：KeyedStream->DataStream。

Interval Join

对指定的时间间隔内使用公共key来连接两个KeyedStream。ProcessJoinFunction<IN1,IN2,OUT>中IN1为第一个DataStream中元素数据类型，IN2为第二个DataStream中的元素数据类型，OUT为结果输出类型。

 keyedStream
            .intervalJoin(keyedStream)
            .between(Time.milliseconds(-2),Time.milliseconds(2))//间隔时间
            .lowerBoundExclusive()//并不包含下限时间
            .upperBoundExclusive()
            .process(new ProcessJoinFunction<String, String, String>() {
                @Override
                public void processElement(String left, String right, Context ctx, Collector<String> out) throws Exception {
                    //...
                }
            });

Iterate

可以使用iterate方法来获取IterativeStream。

IterativeStream<String> iterativeStream = dataStream.iterate();

转换前后的数据类型：DataStream->IterativeStream。

KeyBy

1.介绍

在介绍 KeyBy 函数之前，需要你理解一个概念：KeyedStream。 在实际业务中，我们经常会需要根据数据的某种属性或者单纯某个字段进行分组，然后对不同的组进行不同的处理。举个例子，当我们需要描述一个用户画像时，则需要根据用户的不同行为事件进行加权；再比如，我们在监控双十一的交易大盘时，则需要按照商品的品类进行分组，分别计算销售额。

我们在使用 KeyBy 函数时会把 DataStream 转换成为 KeyedStream，事实上 KeyedStream 继承了 DataStream，KeyedStream 中的元素会根据用户传入的参数进行分组。

在生产环境中使用 KeyBy 函数时要十分注意！该函数会把数据按照用户指定的 key 进行分组，那么相同分组的数据会被分发到一个 subtask 上进行处理，在大数据量和 key 分布不均匀的时非常容易出现数据倾斜和反压，导致任务失败。

常见的解决方式是把所有数据加上随机前后缀，这些我们会在后面的课时中进行深入讲解。

2.例子

逻辑上将数据流元素进行分区，具有相同key的记录将被划分到同一分区。指定Key的方式有多种，这个我们在之前说过了。返回类型KeyedStream<T,KEY>中T代表KeyedStream中元素数据类型，KEY代表虚拟KEY的数据类型。

KeyedStream<String,Tuple> keyedStream = dataStream.keyBy(0);

以下情况的元素不能作为key使用：

1. POJO类型，但没有重写hashCode()，而是依赖Object.hashCode()。
2. 该元素是数组类型。

        KeyedStream<Tuple2<String, Integer>, String> wordAndOneKS = wordAndOneDS.keyBy(
                new KeySelector<Tuple2<String, Integer>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Integer> value) throws Exception {
                        return value.f0;
                    }
                }
        );

keyBy内部使用散列来实现的。
转换前后数据类型：DataStream->KeyedStream。

Map

1.介绍

Map 接受一个元素作为输入，并且根据开发者自定义的逻辑处理后输出。

2.例子

接受一个元素，输出一个元素。MapFunction<T,V>中T代表输入数据类型(map方法的参数类型)，V代表操作结果输出类型(map方法返回数据类型)。

dataStream.map(new MapFunction<String, String>() {
    @Override
    public String map(String line) throws Exception {
        return line.toUpperCase();
    }
});

partitionCustom

自定义分区。使用用户自定义的分区函数对指定key进行分区，partitionCustom只支持单分区。

dataStream.partitionCustom(new Partitioner<String>() {
            
            @Override
            public int partition(String key, int numPartitions) {
                return key.hashCode() % numPartitions;
            }
        },1);

转换前后的数据类型：DataStream->DataStream。

Project

对元组类型的DataStream可以使用Project选取子元组。

DataStream<Tuple2<String,Integer>> dataStream2 = dataStream.project(0,2);

转换前后的数据类型：DataStream->DataStream。

rebalance

以轮询的方式为每个分区均衡分配元素，对于优化数据倾斜该方法非常有效。

dataStream.rebalance();

转换前后的数据类型：DataStream->DataStream。

Reduce

1.介绍

Reduce 函数的原理是，会在每一个分组的 keyedStream 上生效，它会按照用户自定义的聚合逻辑进行分组聚合。

2.例子

对指定的“虚拟”key相同的记录进行滚动合并，也就是当前元素与最后一次的reduce结果进行reduce操作。ReduceFunction<T>中的T代表KeyStream中元素的数据类型。

keyedStream.reduce(new ReduceFunction<String>() {
    @Override
    public String reduce(String value1, String value2) throws Exception {
        return value1 + value2;
    }
});

rescale

根据上下游task数进行分区。

dataStream.rescale();

转换前后的数据类型：DataStream->DataStream。

Select

我们可以对SplitStream分开的流进行选择，可以将其转换成一个或多个DataStream。

splitStream.select("flink");
splitStream.select("other");

shuffle

随机分区。均匀随机将元素进行分区。

dataStream.shuffle();

Split(Deprecated)

我们可以根据某些规则将数据流切分成两个或多个数据流。

dataStream.split(new OutputSelector<String>() {
            @Override
            public Iterable<String> select(String value) {
                List<String> outList = new ArrayList<>();
                if(value.contains("flink"))
                    outList.add("flink");
                else 
                    outList.add("other");
                return outList;
            }
        });

该方法底层引用以被标记为废弃！

转换前后的数据类型：DataStream->SplitStream

Union

联合(Union)两个或多个DataStream，所有DataStream中的元素都会组合成一个新的DataStream。如果联合自身，则每个元素在新的DataStream出现两次。

dataStream.union(dataStream1);

转换前后的数据类型：DataStream*->DataStream。

window

对已经分区的KeyedStream上定义窗口，Window会根据某些规则(比如在最后5s到达的数据)对虚拟“key”相同的记录进行分组。WindowedStream<T, K, W extends Window>中的T为KeyedStream中元素数据类型，K为指定Key的数据类型，W为我们所使用的窗口类型

WindowedStream<String,Tuple,TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)));

转换前后的数据类型：KeyedStream->WindowedStream。

Window Apply

将整个窗口应用在指定函数上，可以对WindowedStream和AllWindowedStream使用。WindowFunction<IN, OUT, KEY, W extends Window>中的IN为输入元素类型，OUT为输出类型元素，KEY为指定的key类型，W为所使用的窗口类型。

windowedStream.apply(new WindowFunction<String, String, Tuple, TimeWindow>() {
    @Override
    public void apply(Tuple tuple, TimeWindow window, Iterable<String> input, Collector<String> out) throws Exception {
        int sumCount = 0;
        for(String line : input){
            sumCount += line.split(" ").length;
        }
        out.collect(String.valueOf(sumCount));
    }
});

AllWindowedStream使用与WindowedStream类似。

转换前后的数据类型：WindowedStream->DataStream或AllWindowedStream->DataStream。

windowAll

我们也可以在常规DataStream上使用窗口，Window根据某些条件(比如最后5s到达的数据)对所有流事件进行分组。AllWindowedStream<T,W extends Window>中的T为DataStream中元素的数据类型，W为我们所使用的窗口类型。

AllWindowedStream<String,TimeWindow> allWindowedStream = dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

注意：该方法在许多时候并不是并行执行的，所有记录都会收集到一个task中

转换前后的数据类型：DataStream->AllWindowedStream。

Window CoGroup

对两个指定的key的DataStream在公共窗口上执行CoGroup，和Join功能类似，但是更加灵活。CoGroupFunction<IN1,IN2,OUT>，IN1代表第一个DataStream中元素类型，IN2代表第二个DataStream中元素类型，OUT为结果输出集合类型。

dataStream
          .coGroup(dataStream1)
          .where(new MyKeySelector()).equalTo(new MyKeySelector())
          .window(TumblingEventTimeWindows.of(Time.seconds(5)))
          .apply(new CoGroupFunction<String, String, String>() {
              @Override
              public void coGroup(Iterable<String> first, Iterable<String> second, Collector<String> out) throws Exception {
 
              }
          });

转换前后的数据类型：DataStream,DataStream->DataStream。

Window Join

在指定key的公共窗口上连接两个数据流。JoinFunction<IN1,IN2,OUT>中的IN1为第一个DataStream中元素数据类型，IN2为第二个DataStream中元素数据类型，OUT为Join结果数据类型。

dataStream
          .join(dataStream1)
          .where(new MyKeySelector()).equalTo(new MyKeySelector())
          .window(TumblingEventTimeWindows.of(Time.seconds(5)))
          .apply(new JoinFunction<String, String, String>() {
              @Override
              public String join(String first, String second) throws Exception {
                  return first + second;
              }
          });

转换前后的数据类型：DataStream,DataStream->DataStream。

Window Reduce/Fold/Aggregation/

对于WindowedStream数据流我们同样也可以应用Reduce、Fold、Aggregation函数。

windowedStream.reduce(new ReduceFunction<String>() {
    @Override
    public String reduce(String value1, String value2) throws Exception {
        return value1 + value2;
    }
});
windowedStream.fold(0, new FoldFunction<String, Integer>() {
    @Override
    public Integer fold(Integer accumulator, String value) throws Exception {
        return accumulator + value.split(" ").length;
    }
});
windowedStream.sum(0);
windowedStream.max("count");

转换前后的数据类型：WindowedStream->DataStream。

（六）结果数据接收器(Data sink)

数据经过Flink处理之后，最终结果会写到file、socket、外部系统或者直接打印出来。数据接收器定义在DataStream类下，我们通过addSink()可以来添加一个接收器。同Source，Flink也提供了一些预定义的Data Sink让我们直接使用。

1.写入文本文件

DataStream提供了两个writeAsText重载方法，写入格式会调用写入对象的toString()方法。

• writeAsText(String path)：将DataStream数据写入到指定文件。
• writeAsText(String path,WriteMode writeMode)：将DataStream数据写入到指定文件，可以通过writeMode来指定如果文件已经存在应该采用什么方式，可以指定OVERWRITE或NO_OVERWRITE。

2.写入CSV文件

DataStream提供了三个写入csv文件的重载方法，对于DataStream中的每个Filed，都会调用其对象的toString()方法作为写入格式。writeAsCsv只能用于元组(Tuple)的DataStream。

writeAsCsv(String path,WriteMode writeMode,String rowDelimiter,String fieldDelimiter)

DataStream提供了两个简易重载方法：

• writeAsCsv(String path)：使用"\n"作为行分隔符，使用","作为列分隔符。
• writeAsCsv(String path,WriteMode writeMode)：使用"\n"作为行分隔符，使用","作为列分隔符。

3.写入Socket

Flink提供了将DataStream作为字节数组写入Socket的方法，通过SerializationSchema来指定输出格式。

writeToSocket(String hostName,int port,SerializationSchema<T> schema)

4.指定输出格式

DataStream提供了自定义文件输出的类和方法，我们能够自定义对象到字节的转换。

writeUsingOutputFormat(OutputFormat<T> format)

5.结果打印

DataStream提供了print和printToErr打印标准输出/标准错误流。DataStream中的每个元素都会调用其toString()方法作为输出格式，我们也可以指定一个前缀字符来区分不同的输出。

• print()：标准输出
• print(String sinkIdentifier)：指定输出前缀
• printToErr()：标准错误输出
• printToErr(String sinkIdentifier)：指定输出前缀

对于并行度大于1的输出，输出结果也将输出任务的标识符作为前缀。

6.自定义输出器

我们一般会自定义输出器，通过实现SinkFunction接口，然后通过DataStream.addSink(sinkFunction)来指定数据接收器。

addSink(SinkFunction<T> sinkFunction)

注意：对于DataStream中的writeXxx()方法一般都是用于测试使用，因为他们并没有参与chaeckpoint，所以它们只有"at-last-once"也就是至少处理一次语义。

如果想要可靠输出，想要使用"exactly-once"语义准确将结果写入到文件系统中，我们需要使用flink-connector-filesystem。此外，我们也可以通过addSink()自定义输出器来使用Flink的checkpoint来完成"exactl-oncey"语义。