<dependencies>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-java</artifactId>
      <version>1.9.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-streaming-java_2.11</artifactId>
      <version>1.9.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.commons</groupId>
      <artifactId>commons-lang3</artifactId>
      <version>3.7</version>
    </dependency>
  </dependencies>

DataStreamSource<String> wordLineDS = env.addSource(new RichSourceFunction<String>() {
      private boolean isCanal = false;
      private String[] words = {
          "important oracle jdk license update",
          "the oracle jdk license has changed for releases starting april 16 2019",
          "the new oracle technology network license agreement for oracle java se is substantially different from prior oracle jdk licenses the new license permits certain uses such as ",
          "personal use and development use at no cost but other uses authorized under prior oracle jdk licenses may no longer be available please review the terms carefully before ",
          "downloading and using this product an faq is available here ",
          "commercial license and support is available with a low cost java se subscription",
          "oracle also provides the latest openjdk release under the open source gpl license at jdk java net"
      };

      @Override
      public void run(SourceContext<String> ctx) throws Exception {
        // 每秒發(fā)送一行文本
        while (!isCanal) {
          int randomIndex = RandomUtils.nextInt(0, words.length);
          ctx.collect(words[randomIndex]);
          Thread.sleep(1000);
        }
      }

      @Override
      public void cancel() {
        isCanal = true;
      }
    });

// 3. 單詞統(tǒng)計
    // 3.1 將文本行切分成一個個的單詞
    SingleOutputStreamOperator<String> wordsDS = wordLineDS.flatMap((String line, Collector<String> ctx) -> {
      // 切分單詞
      Arrays.stream(line.split(" ")).forEach(word -> {
        ctx.collect(word);
      });
    }).returns(Types.STRING);

    //3.2 將單詞轉換為一個個的元組
    SingleOutputStreamOperator<Tuple2<String, Integer>> tupleDS = wordsDS
        .map(word -> Tuple2.of(word, 1))
        .returns(Types.TUPLE(Types.STRING, Types.INT));

    // 3.3 按照單詞進行分組
    KeyedStream<Tuple2<String, Integer>, String> keyedDS = tupleDS.keyBy(tuple -> tuple.f0);

    // 3.4 對每組單詞數(shù)量進行累加
    SingleOutputStreamOperator<Tuple2<String, Integer>> resultDS = keyedDS
        .timeWindow(Time.seconds(3))
        .reduce((t1, t2) -> Tuple2.of(t1.f0, t1.f1 + t2.f1));

    resultDS.print();

public class WordCount {
  public static void main(String[] args) throws Exception {
    // 1. 構建Flink流式初始化環(huán)境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 2. 自定義source - 每秒發(fā)送一行文本
    DataStreamSource<String> wordLineDS = env.addSource(new RichSourceFunction<String>() {
      private boolean isCanal = false;
      private String[] words = {
          "important oracle jdk license update",
          "the oracle jdk license has changed for releases starting april 16 2019",
          "the new oracle technology network license agreement for oracle java se is substantially different from prior oracle jdk licenses the new license permits certain uses such as ",
          "personal use and development use at no cost but other uses authorized under prior oracle jdk licenses may no longer be available please review the terms carefully before ",
          "downloading and using this product an faq is available here ",
          "commercial license and support is available with a low cost java se subscription",
          "oracle also provides the latest openjdk release under the open source gpl license at jdk java net"
      };

      @Override
      public void run(SourceContext<String> ctx) throws Exception {
        // 每秒發(fā)送一行文本
        while (!isCanal) {
          int randomIndex = RandomUtils.nextInt(0, words.length);
          ctx.collect(words[randomIndex]);
          Thread.sleep(1000);
        }
      }

      @Override
      public void cancel() {
        isCanal = true;
      }
    });

    // 3. 單詞統(tǒng)計
    // 3.1 將文本行切分成一個個的單詞
    SingleOutputStreamOperator<String> wordsDS = wordLineDS.flatMap((String line, Collector<String> ctx) -> {
      // 切分單詞
      Arrays.stream(line.split(" ")).forEach(word -> {
        ctx.collect(word);
      });
    }).returns(Types.STRING);

    //3.2 將單詞轉換為一個個的元組
    SingleOutputStreamOperator<Tuple2<String, Integer>> tupleDS = wordsDS
        .map(word -> Tuple2.of(word, 1))
        .returns(Types.TUPLE(Types.STRING, Types.INT));

    // 3.3 按照單詞進行分組
    KeyedStream<Tuple2<String, Integer>, String> keyedDS = tupleDS.keyBy(tuple -> tuple.f0);

    // 3.4 對每組單詞數(shù)量進行累加
    SingleOutputStreamOperator<Tuple2<String, Integer>> resultDS = keyedDS
        .timeWindow(Time.seconds(3))
        .reduce((t1, t2) -> Tuple2.of(t1.f0, t1.f1 + t2.f1));

    resultDS.print();

    env.execute("app");
  }
}

SingleOutputStreamOperator<String> wordsDS = wordLineDS.flatMap((String line, Collector<String> ctx) -> {
      // 切分單詞
      Arrays.stream(line.split(" ")).forEach(word -> {
        ctx.collect(word);
      });
    }).returns(Types.STRING);

@Public
@FunctionalInterface
public interface FlatMapFunction<T, O> extends Function, Serializable {

  /**
  * The core method of the FlatMapFunction. Takes an element from the input data set and transforms
  * it into zero, one, or more elements.
  *
  * @param value The input value.
  * @param out The collector for returning result values.
  *
  * @throws Exception This method may throw exceptions. Throwing an exception will cause the operation
  *          to fail and may trigger recovery.
  */
  void flatMap(T value, Collector<O> out) throws Exception;
}

org.apache.flink.api.common.functions.InvalidTypesException: The generic type parameters of 'Collector' are missing.
  In many cases lambda methods don't provide enough information for automatic type extraction when Java generics are involved.
  An easy workaround is to use an (anonymous) class instead that implements the 'org.apache.flink.api.common.functions.FlatMapFunction' interface.
  Otherwise the type has to be specified explicitly using type information.

SingleOutputStreamOperator<Tuple2<String, Integer>> tupleDS = wordsDS
        .map(word -> Tuple2.of(word, 1))
        .returns(Types.TUPLE(Types.STRING, Types.INT));