目的：用split将一个流分为N个流，用select查询
环境：jdk8 工具：idea

创建maven项目

引入jar包

  1.         <dependency>
  2.             <groupId>org.apache.flink</groupId>
  3.             <artifactId>flink-java</artifactId>
  4.             <version>1.10.1</version>
  5.         </dependency>
  6.         <dependency>
  7.             <groupId>org.apache.flink</groupId>
  8.             <artifactId>flink-streaming-java_2.12</artifactId>
  9.             <version>1.10.1</version>
  10.         </dependency>

代码演示

  1.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2.         env.setParallelism(1);
  3.         DataStream<String> inputStream = env.readTextFile("文件绝对路径");
  6.         DataStream<Test> dataStream = inputStream.map(new MapFunction<String, Test>() {
  7.             @Override
  8.             public Test map(String value) throws Exception {
  9.                 String[] words = value.split(" ");
  10.                 return new Test(words[0], Integer.parseInt(words[1]), words[2]);
  11.             }
  12.         });
  13.         //分流
  14.         SplitStream<Test> spl=dataStream.split(new OutputSelector<Test>() {
  15.             @Override
  16.             public Iterable<String> select(Test value) {
  18.                 return (value.getScore() >= 80) ? Collections.singletonList("high") :
  19.                         Collections.singletonList("low");
  20.             }
  21.         });
  23.         //根据key获取分流
  24.         DataStream<Test> highTempStream = spl.select("high");
  25.         DataStream<Test> lowTempStream = spl.select("low");
  27.         highTempStream.print("highTempStream");
  28.         lowTempStream.print("lowTempStream");
  30.         env.execute();

分流算子.rar(6 KB)

  1. SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
  2. SLF4J: Defaulting to no-operation (NOP) logger implementation
  3. SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
  4. highTempStream> Test{id='1', score=80, content='bs'}
  5. highTempStream> Test{id='2', score=99, content='jj'}
  6. highTempStream> Test{id='3', score=95, content='jw'}
  7. highTempStream> Test{id='1', score=99, content='jj'}