7.Flink数据流API编程指南-上

概述

  Flink中的数据流步伐是在数据流上实现转换的通例步伐(比方，过滤，更新状态，界说窗口，聚合)。数据流由(比方，消息队列、套接字流、文件)创建的。结果通过吸取器返回，比方，吸取器可以将数据写入文件或标准输出(比方下令行终端)。Flink步伐可以在各种上下文中运行，可以独立运行，也可以嵌入到其他步伐中。可以在本地的 jvm 中实验，也可以在多台呆板构成的集群中实验。
什么是 DataStream

  DataStream API 的名称来自用于表现Flink步伐中的数据聚集的特别 DataStream 类。可以将它们视为包罗重复项的不可变数据聚集。这些数据可以是有界的，也可以是无界的，用于处理处罚它们的 API 是雷同的，即流批一体的概念。
  在使用方面，数据流与通例Java Collection雷同，但在一些关键方面有很大差别。它们是不可变的，一旦创建了它们，就不能添加或删除元素。不光遍历此中的元素，还可以使用DataStream API操纵(也称为转换)对它们举行处理处罚。
  可以通过在Flink步伐中添加源来创建初始数据流。然后，可以从中派生新的流，并通过使用map、filter等API方法将它们组合起来。
Flink通例编程规范

  Flink步伐是转换数据流的通例步伐。每个步伐都由雷同的根本部门构成:

• 1,获取实验环境;
  
• 2.加载/创建初始数据，
  
• 3.指定对该数据的转换，
  
• 4.指定把盘算结果存放在那里，
  
• 5.触发步伐实验
  

  现在，将对这些步调举行概述，有关详细信息，请参阅相应的部门。留意，Java DataStream API的全部焦点类都可以在 org.apache.flink.streaming.api 中找到。
  StreamExecutionEnvironment是全部Flink步伐的底子。你可以在StreamExecutionEnvironment上使用这些静态方法得到一个:

1. getExecutionEnvironment();
3. createLocalEnvironment();
5. createRemoteEnvironment(String host, int port, String... jarFiles);

复制代码

  通常，只必要使用 getExecutionEnvironment() ，由于它将根据上下文做出正确选择:如果在IDE中实验步伐或作为通例Java步伐，它将创建一个本地环境，将在本地呆板上实验步伐。如果将步伐打包成一个JAR文件，并通过下令行调用它，那么Flink集群管理器将实验步伐的主方法，getExecutionEnvironment()将返回一个实验环境，用于在集群上实验步伐。
  为了指定命据源，实验环境有几种方法可以使用各种方法从文件中读取:可以将它们作为CSV文件逐行读取，也可以使用任何其他提供的源。要将文本文件作为行序列读取，可以使用:

1. final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
3. DataStream<String> text = env.readTextFile("file:///path/to/file");

复制代码

  这将提供一个数据流，然后可以在其上应用转换以创建新的派生数据流。
通过调用带有转换函数的DataStream上的方法来应用转换。比方，一个map应用如下:

1. DataStream<String> input = ...;
3. DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {
4.     @Override
5.     public Integer map(String value) {
6.         return Integer.parseInt(value);
7.     }
8. });

复制代码

  这将原始聚集中的每个String转换为Integer来创建一个新的数据流。
一旦有了包罗最闭幕果的数据流，就可以通过创建吸取器将其写入外部体系。这些只是一些创建吸取器的示例方法:

1. writeAsText(String path);
3. print();

复制代码

  上面的步调完成时，就必要通过调用StreamExecutionEnvironment上的execute()来触发步伐实验。根据ExecutionEnvironment的范例，在自己的呆板上实验，大概将步伐提交到集群上实验。
  execute()方法将期待作业完成，然后返回jobeexecutionresult，此中包罗实验时间和累加器结果。
如果不想期待作业完成，可以通过调用StreamExecutionEnvironment上的executeAsync()来触发异步作业实验。它将返回一个JobClient，可以使用它与刚刚提交的作业举行通讯。比方，下面是怎样通过使用executeAsync()来实现execute()的语义。

1. final JobClient jobClient = env.executeAsync();
3. final JobExecutionResult jobExecutionResult = jobClient.getJobExecutionResult().get();

复制代码

  关于步伐实验的末了一部门对于明白何时以及怎样实验Flink操纵至关紧张。全部Flink步伐都是惰性实验的:当步伐的main方法被实验时，数据加载和转换不会直接发生。相反，每个操纵都被创建并添加到数据流图中。当实验环境中的execute()调用显式触发实验时，操纵才会实际实验。步伐是在本地实验还是在集群上实验取决于实验环境的范例。
  惰性盘算可以构建复杂的步伐，Flink将这些步伐作为一个团体规划单元实验。
编程案例

  下面是一个Flink批处理处罚案例，代表了Flink流批处理处罚的根本流程。

1. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
2. import org.apache.flink.streaming.api.datastream.DataStream;
3. import org.apache.flink.api.common.functions.FilterFunction;
5. public class Example {
7.     public static void main(String[] args) throws Exception {
8.         final StreamExecutionEnvironment env =
9.                 StreamExecutionEnvironment.getExecutionEnvironment();
11.         DataStream<Person> flintstones = env.fromElements(
12.                 new Person("Fred", 35),
13.                 new Person("Wilma", 35),
14.                 new Person("Pebbles", 2));
16.         DataStream<Person> adults = flintstones.filter(new FilterFunction<Person>() {
17.             @Override
18.             public boolean filter(Person person) throws Exception {
19.                 return person.age >= 18;
20.             }
21.         });
23.         adults.print();
25.         env.execute();
26.     }
28.     public static class Person {
29.         public String name;
30.         public Integer age;
32.         public Person(String name, Integer age) {
33.             this.name = name;
34.             this.age = age;
35.         }
37.         public String toString() {
38.             return this.name + ": age " + this.age.toString();
39.         }
40.     }
41. }

复制代码

数据源

  源是步伐读取输入的地方。可以使用 StreamExecutionEnvironment.addSource(sourceFunction) 将源附加到步伐中。Flink附带了很多预实现的源代码函数，可以通过为非并行源代码实现 SourceFunction ，大概通过实现 ParallelSourceFunction 接口或为并行源代码扩展 RichParallelSourceFunction 来编写自己的自界说源代码。
  有几个预界说的流源可以从 StreamExecutionEnvironment 中访问
基于文件

• readTextFile(path) -逐行读取文本文件，即服从TextInputFormat规范的文件，并将其作为字符串返回。
  
• readFile(fileInputFormat, path) -按照指定的文件输入格式读取(一次)文件。
  
• readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) -这是前两个方法在内部调用的方法。它根据给定的fileInputFormat读取路径中的文件。根据所提供的watchType，这个源可以定期(每隔毫秒)监视新数据的路径(fileprocessingmode . process_continuous)，大概处理处罚一次当前路径中的数据并退出(FileProcessingMode.PROCESS_ONCE)。使用pathFilter，用户可以进一步打扫正在处理处罚的文件。
  

Socket-based

socketTextStream - 从 socket 中读取，元素之间使用分隔器
Collection-based:

fromCollection(Collection) - 通过Java Java.util.Collection创建数据流，全部聚集中的元素是同一种范例。
fromCollection(Iterator, Class) - 通过iterator创建数据流，通过 Class 指定返回 iterator 返回数据的范例
fromElements(T …) - Creates a data stream from the given sequence of objects. All objects must be of the same type.
fromParallelCollection(SplittableIterator, Class) - Creates a data stream from an iterator, in parallel. The class specifies the data type of the elements returned by the iterator.
fromSequence(from, to) - Generates the sequence of numbers in the given interval, in parallel.
自界说

竣事

  下部门在下一篇中增补。