目的：读取集合中的数据
环境：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>

代码演示

DataStream用于实时数据统计

  1.   //创建流处理执行环境
  2.   StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
  4.   List<Test> testData = Arrays.asList(
  5.                 new Test(1, "测试1", "测试1"),
  6.                 new Test(2, "测试2", "测试2"),
  7.                 new Test(3, "测试3", "测试3"));
  8.   DataStream<Test> dataStream = env.fromCollection(testData);
  10.   dataStream.print();
  12.   env.execute();   //如果要保持顺序可以设置全局并行度为1

集合中读取数据.rar(4 KB)

Test类

  1. package com.hello.flink;
  3. public class Test {
  4.     private Integer id;
  5.     private String name;
  6.     private String content;
  8.     public Integer getId() {
  9.         return id;
  10.     }
  12.     public void setId(Integer id) {
  13.         this.id = id;
  14.     }
  16.     public String getName() {
  17.         return name;
  18.     }
  20.     public void setName(String name) {
  21.         this.name = name;
  22.     }
  24.     public String getContent() {
  25.         return content;
  26.     }
  28.     public void setContent(String content) {
  29.         this.content = content;
  30.     }
  32.     public Test(Integer id, String name, String content) {
  33.         this.id = id;
  34.         this.name = name;
  35.         this.content = content;
  36.     }
  38. }

结果

  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. 4> Test{id=1, name='测试1', content='测试1'}
  5. 2> Test{id=3, name='测试3', content='测试3'}
  6. 1> Test{id=2, name='测试2', content='测试2'}