获取执行环境（execution environment）
读取数据源（source）
定义基于数据的转换操作（transformations）
定义计算结果的输出位置（sink）
触发程序执行（execute）

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();

StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment
    .createRemoteEnvironment(
        "host", // JobManager主机名
        1234, // JobManager进程端口号
      "path/to/jarFile.jar"// 提交给JobManager的JAR包
      );

// 批处理环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

bin/flink run -Dexecution.runtime-mode=BATCH ...
在提交作业时，增加execution.runtime-mode参数，指定值为BATCH。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

env.execute();

DataStream<String> stream = env.addSource(...);

import java.sql.Timestamp;
publicclass Event {
  public String user;
  public String url;
  public Long timestamp;
  public Event() {
  }
  public Event(String user, String url, Long timestamp) {
    this.user = user;
    this.url = url;
    this.timestamp = timestamp;
  }
  @Override
  public String toString() {
    return"Event{" +
        "user='" + user + '\'' +
        ", url='" + url + '\'' +
        ", timestamp=" + new Timestamp(timestamp) +
         '}';
  }
}

类是公有（public）的
 有一个无参的构造方法
 所有属性都是公有（public）的
 所有属性的类型都是可以序列化的

public static void main(String[] args) throws Exception {
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.setParallelism(1);
  ArrayList<Event> clicks = new ArrayList<>();
  clicks.add(new Event("Mary","./home",1000L));
  clicks.add(new Event("Bob","./cart",2000L));
  DataStream<Event> stream = env.fromCollection(clicks);
  stream.print();
  env.execute();
}

DataStreamSource<Event> stream2 = env.fromElements(
   new Event("Mary", "./home", 1000L),
   new Event("Bob", "./cart", 2000L)
);

DataStream<String> stream = env.readTextFile("clicks.csv");

参数可以是目录，也可以是文件；还可以从hdfs目录下读取，使用路径hdfs://...；
 路径可以是相对路径，也可以是绝对路径；
 相对路径是从系统属性user.dir获取路径: idea下是project的根目录, standalone模式下是集群节点根目录；

DataStream<String> stream = env.socketTextStream("localhost", 7777);

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>

publicclass SourceKafka {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    Properties properties = new Properties();
    properties.setProperty("bootstrap.servers", "hadoop102:9092");
    properties.setProperty("group.id", "consumer-group");
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.setProperty("auto.offset.reset", "latest");
    DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>(
        "clicks",
        new SimpleStringSchema(),
        properties
    ));
    stream.print("Kafka");
    env.execute();
  }
}

第一个参数topic，定义了从哪些主题中读取数据。可以是一个topic，也可以是topic列表，还可以是匹配所有想要读取的topic的正则表达式。当从多个topic中读取数据时，Kafka连接器将会处理所有topic的分区，将这些分区的数据放到一条流中去。
第二个参数是一个DeserializationSchema或者KeyedDeserializationSchema。Kafka消息被存储为原始的字节数据，所以需要反序列化成Java或者Scala对象。上面代码中使用的SimpleStringSchema，是一个内置的DeserializationSchema，它只是将字节数组简单地反序列化成字符串。DeserializationSchema和KeyedDeserializationSchema是公共接口，所以我们也可以自定义反序列化逻辑。
第三个参数是一个Properties对象，设置了Kafka客户端的一些属性。

run()方法：通过运行时上下文对象SourcContext循环生成数据，并发送到流中；
cancel()方法：通过标识位控制退出循环，来达到中断数据源的效果。

publicclass ClickSource implements SourceFunction<Event> {
   private Boolean running = true;
   @Override
   public void run(SourceContext<Event> sourceContext) throws Exception {
     Random random = new Random();
     String[] users = {"Mary", "Bob"};
     String[] urls = {"./home", "./cart", “./fav”, “./prod?id=1”, “./prod?id=2”};
     while (running) {
       sourceContext.collect(new Event(
           users[random.nextInt(users.length)],
           urls[random.nextInt(urls.length)],
           Calendar.getInstance().getTimeInMillis()
       ));
     }
   }
   @Override
   public void cancel() {
     running = false;
   }
 }

env.addSource(new ClickSource())

publicclass SourceCustom {
public static void main(String[] args) throws Exception {
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.setParallelism(1);
    //有了自定义的source function，调用addSource方法
  DataStreamSource<Event> stream = env.addSource(new ClickSource());
  stream.print("SourceCustom");
  env.execute();
}
}

Java元组类型（TUPLE）：这是Flink内置的元组类型，是Java API的一部分。最多25个字段，也就是从Tuple0~Tuple25，不支持空字段
Scala 样例类及Scala元组：不支持空字段
行类型（ROW）：可以认为是具有任意个字段的元组,并支持空字段
POJO：Flink自定义的类似于Java bean模式的类

Flink支持所有的Java类和Scala类。不过如果没有按照上面POJO类型的要求来定义，就会被Flink当作泛型类来处理。Flink会把泛型类型当作黑盒，无法获取它们内部的属性；它们也不是由Flink本身序列化的，而是由Kryo序列化的。
在这些类型中，元组类型和POJO类型最为灵活，因为它们支持创建复杂类型。而相比之下，POJO还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。所以，在项目实践中，往往会将流处理程序中的元素类型定为Flink的POJO类型。

类是公共的（public）和独立的（standalone，也就是说没有非静态的内部类）；
 类有一个公共的无参构造方法；
 类中的所有字段是public且非final的；或者有一个公共的getter和setter方法，这些方法需要符合Java bean的命名规范。
所以我们看到，之前的UserBehavior，就是我们创建的符合Flink POJO定义的数据类型。

.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));

returns(new TypeHint<Tuple2<Integer, SomeType>>(){})