190728-flink官方文档阅读和实战记录

三、应用开发

地址：https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/projectsetup/java_api_quickstart.html

（一）项目构建设置

1、Java项目模板

在D:GiteeProject目录输入命令：

mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.8.0

然后会自动生成一个示例项目。

示例项目是一个Maven项目，它包含两个类：StreamingJob和BatchJob是DataStream和DataSet程序的基本框架程序。

一、概念

（一）教程

1、API教程

（1）Data Stream API

地址：https://ci.apache.org/projects/flink/flink-docs-release-1.8/tutorials/datastream_api.html

在flink-demo项目下建立子项目wiki-edits，引入下列依赖：

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-wikiedits_2.11</artifactId>
        <version>${flink.version}</version>
    </dependency>
</dependencies>

新建类WikipediaAnalysis，步骤如下：

Flink程序的第一步是创建一个StreamExecutionEnvironment （或者ExecutionEnvironment如果您正在编写批处理作业）。这可用于设置执行参数并创建从外部系统读取的源。所以让我们继续把它添加到main方法：

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

接下来，我们将创建一个从Wikipedia IRC日志中读取的源：

DataStream<WikipediaEditEvent> edits = see.addSource(new WikipediaEditsSource());

这创建了一个我们可以进一步处理DataStream的WikipediaEditEvent元素。出于本示例的目的，我们感兴趣的是确定每个用户在特定时间窗口中添加或删除的字节数，比如说五秒。为此，我们首先要指定我们要在用户名上键入流，也就是说此流上的操作应考虑用户名。在我们的例子中，窗口中编辑的字节的总和应该是每个唯一的用户。对于键入流，我们必须提供一个KeySelector，如下所示：

KeyedStream<WikipediaEditEvent, String> keyedEdits = edits
    .keyBy(new KeySelector<WikipediaEditEvent, String>() {
        @Override
        public String getKey(WikipediaEditEvent event) {
            return event.getUser();
        }
    });

这为我们提供了一个WikipediaEditEvent具有String密钥的用户名。我们现在可以指定我们希望在此流上加上窗口，并根据这些窗口中的元素计算结果。窗口指定要在其上执行计算的Stream的切片。在无限的元素流上计算聚合时需要Windows。在我们的例子中，我们将说我们想要每五秒聚合一次编辑的字节总和：

DataStream<Tuple2<String, Long>> result = keyedEdits
    .timeWindow(Time.seconds(5))
    .fold(new Tuple2<>("", 0L), new FoldFunction<WikipediaEditEvent, Tuple2<String, Long>>() {
        @Override
        public Tuple2<String, Long> fold(Tuple2<String, Long> acc, WikipediaEditEvent event) {
            acc.f0 = event.getUser();
            acc.f1 += event.getByteDiff();
            return acc;
        }
    });

第一个调用，.timeWindow()指定我们希望有五秒钟的翻滚（非重叠）窗口。第二个调用为每个唯一键指定每个窗口切片的折叠变换。在我们的例子中，我们从一个初始值开始，("", 0L)并在该时间窗口中为用户添加每个编辑的字节差异。生成的Stream现在包含Tuple2<String, Long>每五秒钟发出一次的用户。

剩下要做的就是将流打印到控制台并开始执行：

result.print();

see.execute();

最后一次调用是启动实际Flink工作所必需的。所有操作（例如创建源，转换和接收器）仅构建内部操作的图形。只有在execute()被调用时 才会在集群上抛出或在本地计算机上执行此操作图。

综上，类WikipediaAnalysis的代码如下：

import org.apache.flink.api.common.functions.FoldFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditEvent;
import org.apache.flink.streaming.connectors.wikiedits.WikipediaEditsSource;

/**
 * Created by zhangyuxuan on 2019/7/28/028 22:07.
 */
public class WikipediaAnalysis {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<WikipediaEditEvent> edits = see.addSource(new WikipediaEditsSource());

        KeyedStream<WikipediaEditEvent, String> keyedEdits = edits
                .keyBy(new KeySelector<WikipediaEditEvent, String>() {
                    @Override
                    public String getKey(WikipediaEditEvent event) {
                        return event.getUser();
                    }
                });

        DataStream<Tuple2<String, Long>> result = keyedEdits
                .timeWindow(Time.seconds(5))
                .fold(new Tuple2<>("", 0L), new FoldFunction<WikipediaEditEvent, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> fold(Tuple2<String, Long> acc, WikipediaEditEvent event) {
                        acc.f0 = event.getUser();
                        acc.f1 += event.getByteDiff();
                        return acc;
                    }
                });

        result.print();

        see.execute();
    }
}