publicclass MyFlatMap extends RichFlatMapFunction<IN, OUT>> {
@Override
public void open(Configuration configuration) {
  // 做一些初始化工作
  // 例如建立一个和MySQL的连接
}
@Override
public void flatMap(IN in, Collector<OUT out) {
  // 对数据库进行读写
}
@Override
public void close() {
  // 清理工作，关闭和MySQL数据库的连接。
}
}

publicclass ShuffleExample {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4).setParallelism(1);
    stream.shuffle().print().setParallelism(2);
    env.execute();
  }
}

stream.addSink(newSinkFunction(…));

行编码：StreamingFileSink.forRowFormat（basePath，rowEncoder）。
 批量编码：StreamingFileSink.forBulkFormat（basePath，bulkWriterFactory）。

publicclass SinkFile {
 public static void main(String[] args) throws Exception {
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   env.setParallelism(1);
   DataStream<String> input = env.fromElements("hello world", "hello flink");
   final StreamingFileSink<String> sink = StreamingFileSink
       .forRowFormat(new Path("./output"), 
new SimpleStringEncoder<String>("UTF-8")).withRollingPolicy(
           DefaultRollingPolicy.builder()
             .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
             .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
             .withMaxPartSize(1024 * 1024 * 1024)
             .build())
       .build();
   input.addSink(sink);
   env.execute();
 }
}

至少包含15分钟的数据
最近5分钟没有收到新的数据
文件大小已达到1 GB

publicclass SinkKafka {
 public static void main(String[] args) throws Exception {
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   env.setParallelism(1);
   Properties properties = new Properties();
   properties.put("bootstrap.servers", "hadoop102:9092");
   DataStreamSource<String> stream = env.readTextFile("input/clicks.csv");
   stream
       .addSink(new FlinkKafkaProducer<String>(
           "clicks",
           new SimpleStringSchema(),
           properties
       ));
   env.execute();
 }
}

bin/kafka-console-consumer.sh --bootstrap-server hadoop100:9092 --topic clicks

<dependency>
 <groupId>org.apache.bahir</groupId>
 <artifactId>flink-connector-redis_2.11</artifactId>
 <version>1.0</version>
</dependency>

publicclass SinkRedis {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("hadoop200").build();
    env.addSource(new ClickSource())
        .addSink(new RedisSink<Event>(conf, new MyRedisMapper()));
    env.execute();
  }
}

JFlinkJedisConfigBase：Jedis的连接配置
 RedisMapper：Redis映射类接口，说明怎样将数据转换成可以写入Redis的类型

publicstaticclass MyRedisMapper implements RedisMapper<Event> {
  @Override
  public String getKeyFromData(Event e) {
    return e.user;
  }
  @Override
  public String getValueFromData(Event e) {
    return e.url;
  }
  @Override
  public RedisCommandDescription getCommandDescription() {
    returnnew RedisCommandDescription(RedisCommand.HSET, "clicks");
  }
}

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

publicclass SinkToEs {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    ArrayList<HttpHost> httpHosts = new ArrayList<>();
    httpHosts.add(new HttpHost("hadoop102", 9200, "http"));
    ElasticsearchSink.Builder<Event> esBuilder = new ElasticsearchSink.Builder<>(
        httpHosts,
        new ElasticsearchSinkFunction<Event>() {
          @Override
          public void process(Event event, RuntimeContext runtimeContext, RequestIndexer requestIndexer) {
            HashMap<String, String> data = new HashMap<>();
            data.put(event.user, event.url);
            IndexRequest indexRequest = Requests
                .indexRequest()
                .index("clicks")
               .type("type")
                .source(data);
            requestIndexer.add(indexRequest);
          }
        }
    );
    DataStream<Event> stream = env.fromElements(new Event("Mary", "./home", 1000L));
    stream.addSink(esBuilder.build());
    env.execute();
  }
}

httpHosts：连接到的Elasticsearch集群主机列表
 elasticsearchSinkFunction：这并不是我们所说的SinkFunction，而是用来说明具体处理逻辑、准备数据向Elasticsearch发送请求的函数

<dependency>
 <groupId>org.apache.flink</groupId>
 <artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
 <version>${flink.version}</version>
</dependency>
<dependency>
 <groupId>mysql</groupId>
 <artifactId>mysql-connector-java</artifactId>
 <version>5.1.47</version>
</dependency>

mysql> createtable clicks(
  -> uservarchar(20) notnull,
  -> urlvarchar(100) notnull);

publicclass SinkToMySQL {
public static void main(String[] args) throws Exception {
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.setParallelism(1);
  env
      .fromElements(
          new Event("Mary", "./home", 1000L),
          new Event("Bob", "./cart", 2000L)
      )
      .addSink(
          JdbcSink.sink(
              "INSERT INTO clicks (user, url) VALUES (?, ?)",
              (statement, r) -> {
                statement.setString(1, r.user);
                statement.setString(2, r.url);
              },
              JdbcExecutionOptions.builder()
                  .withBatchSize(1000)
                  .withBatchIntervalMs(200)
                  .withMaxRetries(5)
                  .build(),
              new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                  .withUrl("jdbc:mysql://localhost:3306/test")
                  // 使用MySQL 5.7的话，没有cj
                  .withDriverName("com.mysql.cj.jdbc.Driver")
                  .withUsername("username")
                  .withPassword("password")
                  .build()
          )
      );
  env.execute();
}
}

stream.addSink(new MySinkFunction<String>());

事件时间，是指每个事件在对应的设备上发生的时间，也就是数据生成的时间。
数据一旦产生，这个时间自然就确定了，所以它可以作为一个属性嵌入到数据中。这其实就是这条数据记录的“时间戳”（Timestamp）。
在事件时间语义下，我们对于时间的衡量，就不看任何机器的系统时间了，而是依赖于数据本身。由于分布式系统中网络传输延迟的不确定性，实际应用中我们要面对的数据流往往是乱序的。在这种情况下，就不能简单地把数据自带的时间戳当作时钟了，而需要用另外的标志来表示事件时间进展，在Flink中把它叫作事件时间的“水位线”（Watermarks）。

水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据
水位线主要的内容是一个时间戳，用来表示当前事件时间的进展
水位线是基于数据的时间戳生成的
水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进
水位线可以通过设置延迟，来保证正确处理乱序数据
一个水位线Watermark(t)，表示在当前流中事件时间已经达到了时间戳t, 这代表t之前的所有数据都到齐了，之后流中不会出现时间戳t’ ≤ t的数据
水位线是Flink流处理中保证结果正确性的核心机制，它往往会跟窗口一起配合，完成对乱序数据的正确处理。

public SingleOutputStreamOperator<T> assignTimestampsAndWatermarks(
WatermarkStrategy<T> watermarkStrategy)

DataStream<Event> stream = env.addSource(new ClickSource());
DataStream<Event> withTimestampsAndWatermarks = 
stream.assignTimestampsAndWatermarks(*<watermark strategy>*);

publicinterface WatermarkStrategy<T> 
  extends TimestampAssignerSupplier<T>,
      WatermarkGeneratorSupplier<T>{
  @Override
  TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);
  @Override
  WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}

TimestampAssigner：主要负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。
WatermarkGenerator：主要负责按照既定的方式，基于时间戳生成水位线。
在WatermarkGenerator接口中，主要又有两个方法：onEvent()和onPeriodicEmit()。
onEvent：每个事件（数据）到来都会调用的方法，它的参数有当前事件、时间戳，以及允许发出水位线的一个WatermarkOutput，可以基于事件做各种操作
onPeriodicEmit：周期性调用的方法，可以由WatermarkOutput发出水位线。周期时间为处理时间，可以调用环境配置的.setAutoWatermarkInterval()方法来设置，默认为200ms。
env.getConfig().setAutoWatermarkInterval(60 * 1000L);

stream.assignTimestampsAndWatermarks(
    WatermarkStrategy.<Event>forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
          @Override
          public long extractTimestamp(Event element, long recordTimestamp) {
            return element.timestamp;
          }
        })
);

publicclass WatermarkExample {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env
        .addSource(new ClickSource())
        // 插入水位线的逻辑
        .assignTimestampsAndWatermarks(
             // 针对乱序流插入水位线，延迟时间设置为5s
              WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
       .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                  // 抽取时间戳的逻辑
                  @Override
                  public long extractTimestamp(Event element, long recordTimestamp) {
                    return element.timestamp;
                  }
                })
        )
        .print();
    env.execute();
  }
}

import com.atguigu.bean.Event;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
// 自定义水位线的产生
publicclass CustomPeriodicWatermarkExample {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env
        .addSource(new ClickSource())
        .assignTimestampsAndWatermarks(new CustomWatermarkStrategy())
        .print();
    env.execute();
  }
  publicstaticclass CustomWatermarkStrategy implements WatermarkStrategy<Event> {
    @Override
    public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
      returnnew SerializableTimestampAssigner<Event>() {
        @Override
        public long extractTimestamp(Event element, long recordTimestamp) {
          return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段
        }
      };
    }
    @Override
    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
      returnnew CustomBoundedOutOfOrdernessGenerator();
    }
  }
  publicstaticclass CustomBoundedOutOfOrdernessGenerator implements WatermarkGenerator<Event> {
    private Long delayTime = 5000L; // 延迟时间
    private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳
    @Override
    public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
      // 每来一条数据就调用一次
      maxTs = Math.max(event.timestamp, maxTs); // 更新最大时间戳
    }
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
      // 发射水位线，默认200ms调用一次
      output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
    }
  }
}

publicclass PunctuatedGenerator implements WatermarkGenerator<Event> {
  @Override
  public void onEvent(Event r, long eventTimestamp, WatermarkOutput output) {
// 只有在遇到特定的itemId时，才发出水位线
    if (r.user.equals("Mary")) {
      output.emitWatermark(new Watermark(r.timestamp - 1));
    }
  }
  @Override
  public void onPeriodicEmit(WatermarkOutput output) {
    // 不需要做任何事情，因为我们在onEvent方法中发射了水位线
  }
}

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import java.sql.Timestamp;
import java.util.Calendar;
import java.util.Random;
publicclass EmitWatermarkInSourceFunction {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env.addSource(new ClickSource()).print();
    env.execute();
  }
  // 泛型是数据源中的类型
  publicstaticclass ClickSource implements SourceFunction<Event> {
    privateboolean running = true;
    @Override
    public void run(SourceContext<Event> sourceContext) throws Exception {
      Random random = new Random();
      String[] userArr = {"Mary", "Bob", "Alice"};
      String[] urlArr  = {"./home", "./cart", "./prod?id=1"};
      while (running) {
        long currTs = Calendar.getInstance().getTimeInMillis(); // 毫秒时间戳
        String username = userArr[random.nextInt(userArr.length)];
        String url    = urlArr[random.nextInt(urlArr.length)];
        Event event = new Event(username, url, currTs);
        // 使用collectWithTimestamp方法将数据发送出去，并指明数据中的时间戳的字段
        sourceContext.collectWithTimestamp(event, event.timestamp);
        // 发送水位线
        sourceContext.emitWatermark(new Watermark(event.timestamp - 1L));
        Thread.sleep(1000L);
      }
    }
    @Override
    public void cancel() {
      running = false;
    }
  }
}

（1）上游并行子任务发来不同的水位线，当前任务会为每一个分区设置一个“分区水位线”（Partition Watermark），这是一个分区时钟；而当前任务自己的时钟，就是所有分区时钟里最小的那个。
（2）当有一个新的水位线（第一分区的4）从上游传来时，当前任务会首先更新对应的分区时钟；然后再次判断所有分区时钟中的最小值，如果比之前大，说明事件时间有了进展，当前任务的时钟也就可以更新了。这里要注意，更新后的任务时钟，并不一定是新来的那个分区水位线，比如这里改变的是第一分区的时钟，但最小的分区时钟是第三分区的3，于是当前任务时钟就推进到了3。当时钟有进展时，当前任务就会将自己的时钟以水位线的形式，广播给下游所有子任务。
（3）再次收到新的水位线（第二分区的7）后，执行同样的处理流程。首先将第二个分区时钟更新为7，然后比较所有分区时钟；发现最小值没有变化，那么当前任务的时钟也不变，也不会向下游任务发出水位线。
（4）同样道理，当又一次收到新的水位线（第三分区的6）之后，第三个分区时钟更新为6，同时所有分区时钟最小值变成了第一分区的4，所以当前任务的时钟推进到4，并发出时间戳为4的水位线，广播到下游各个分区任务。
水位线在上下游任务之间的传递，非常巧妙地避免了分布式系统中没有统一时钟的问题，每个任务都以“处理完