flink job 提交源码 flink源码解读

本篇先介绍流中的数据，再通过一个完整的案例，介绍流处理应用中的几个重要组成部分。

流里面是什么

Flink的DataStream API可以基于Java或Scala编写，流内部支持很多种类型，比如Java的基础类型，如String、Long、Integer、Boolean、Array等；也支持复杂类型，如Tuples、POJOS、Scala Case class等。也支持使用Kryo或Avro进行序列化。Flink原生的序列化在tuple和pojo上效率更高，对于java flink定义了自己的tuple，最多支持25个列。

Tuple2<String, Integer> person = Tuple2.of("Fred", 35);

// 从0开始索引 
String name = person.f0;
Integer age = person.f1;

也能自动识别POJO类型，但需要满足：类是public且不是静态内部类，类包含public的无参构造方法，所有非静态或非transient字段都是public修饰或包含public的getter setter方法。比如：

public class Person {
public String name;  
public Integer age;  
public Person() {};  
public Person(String name, Integer age) {  
        . . .
    };  
}  

Person person = new Person("Fred Flintstone", 35);

完整样例

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.functions.FilterFunction;

public class Example {

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

        DataStream flintstones = env.fromElements(new Person("Fred", 35),new Person("Wilma", 35),new Person("Pebbles", 2));
        DataStream adults = flintstones.filter(new FilterFunction() {@Overridepublic boolean filter(Person person) throws Exception {return person.age >= 18;
            }
        });
        adults.print();
        env.execute();
    }public static class Person {public String name;public Integer age;public Person() {};public Person(String name, Integer age) {this.name = name;this.age = age;
        };public String toString() {return this.name.toString() + ": age " + this.age.toString();
        };
    }
}

执行环境

每个Flink应用都需要一个执行环境，比如上面代码中的env，当调用env.execute() 时，背后会执行job图的编译，并发送到JM。在JM中把job图在编译成执行图，描述并行的执行过程，然后交由TM执行。每个算子的子任务会在task slot中并行执行。注意，如果不调用env.execute()任务就不会触发执行。

流的输入

上面的例子中，使用env.fromElements获取数据流，也可以使用fromCollection。如:

List people = new ArrayList();
people.add(new Person("Fred", 35));
people.add(new Person("Wilma", 35));
people.add(new Person("Pebbles", 2));
DataStream flintstones = env.fromCollection(people);

另一种简单的方式是通过socket获取数据:

DataStream<String> lines = env.socketTextStream("localhost", 9999)

也可以从文件中读取：

DataStream lines = env.readTextFile("file:///path");

在具体的应用中，最常见的数据源需要支持低延迟、高吞吐、支持回放，这样才能保证高性能和高可用，比如Kafka。

流的输出

在上面的例子中, adults.print会在taskmanager中输出日志，如果使用IDE会在控制台输出。在print中会把每条记录调用toString方法输出打印。如：

1> Fred: age 35
2> Wilma: age 35

前面的1>和2>代表具体的子任务(可能是不同的线程)。在生产环境，一般会使用StreamingFileSink，不同的数据库或者发布订阅系统。

调试

在生产环境，应用会运行在远程集群或容器中。JM和TM日志在失败时会很有用，但是如果在IDE中断点调试会更方便点。

总结

到此你可以开始编写flink的代码，并运行一个简单的应用流了。