Source

InputFormat/RichInputFormat接口
CsvInputFormat TextInputFormat
ExecutionEnvironment



// 读取本地文件
DataSet<String> localLines = env.readTextFile("file:///path/file");
// 读取HDSF文件
DataSet<String> hdfsLines = env.readTextFile("hdfs://host:port/path/file");


DataSet<StringValue> localLines = env.readTextFileWithValue("file:///some/local/file");
DataSet<StringValue> hdfsLines = env.readTextFileWithValue("hdfs://host:port/file/path");

StringValue 是一种可变的String类型


// read a CSV file with five fields, taking only two of them
// 读取一个具有5个字段的CSV文件，只取第一个和第四个字段
DataSet<Tuple2<String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
                               .includeFields("10010")  
                              .types(String.class, Double.class);

// 读取一个有三个字段的CSV文件，将其转为POJO类型
DataSet<Person>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")
                         .pojoType(Person.class, "name", "age", "zipcode");


DataSet<String> Data = env.readFileOfPrimitives("file:///some/local/file", String.class);

DataSet<String> data = env.fromCollection(arrayList);

DataSet<String> stringDataSource = env.fromElements("hello Flink What is Apache Flink");

DataSet<Long> longDataSource = env.generateSequence(1, 20);


env.readFile(new MyInputFormat(), "file:///some/local/file");


DataSet<Tuple2<String, Integer> dbData =
    env.createInput(
      JDBCInputFormat.buildJDBCInputFormat()
                     .setDrivername("com.mysql.jdbc.Driver")
                     .setDBUrl("jdbc:mysql://localhost/mydb")
                     .setQuery("select name, age from stu")
                     .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO))
                     .finish()
    );