• 从集合中读数据方式一

package test01;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

public class TestRead {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> source = env.fromElements(2, 142, 664);

        source.print();
        env.execute();
    }
}

• 从集合中读数据方式二

• package test01;
  
  import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  import org.apache.flink.connector.file.src.FileSource;
  import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
  import org.apache.flink.core.fs.Path;
  import org.apache.flink.streaming.api.datastream.DataStreamSource;
  import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  
  import java.util.Arrays;
  
  public class TestRead {
  
      public static void main(String[] args) throws Exception {
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          DataStreamSource<Integer> source = env.fromCollection(Arrays.asList(5,523,637));
  
          source.print();
          env.execute();
      }
  }
  

   
• 从文件中读数据 

需要先引入依赖：

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

package test01;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

public class TestRead {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //DataStreamSource<Integer> source = env.fromElements(2, 142, 664);

        FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                new TextLineInputFormat(),
                new Path("test_input/test_word.txt")
        ).build();
        //此处参数三自定义一个suorceName即可“filesource”
        env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "filesource").print();
        env.execute();
    }
}

运行结果：