Flink DataSet API编程指南

r囧r小猫 2021-11-22 06:42 340阅读 0赞

Flink中的DataSet程序是实现数据集转换的常规程序（例如，过滤，映射，连接，分组）。数据集最初是从某些来源创建的（例如，通过读取文件或从本地集合创建）。结果通过接收器返回，接收器可以例如将数据写入（分布式）文件或标准输出（例如命令行终端）。Flink程序可以在各种环境中运行，独立运行或嵌入其他程序中。执行可以在本地JVM中执行，也可以在许多计算机的集群上执行。

public class WordCountExample {	
        public static void main(String[] args) throws Exception {	
            final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();	
    	
            DataSet<String> text = env.fromElements(	
                "Who's there?",	
                "I think I hear them. Stand, ho! Who's there?");	
    	
            DataSet<Tuple2<String, Integer>> wordCounts = text	
                .flatMap(new LineSplitter())	
                .groupBy(0)	
                .sum(1);	
    	
            wordCounts.print();	
        }	
    	
        public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {	
            @Override	
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {	
                for (String word : line.split(" ")) {	
                    out.collect(new Tuple2<String, Integer>(word, 1));	
                }	
            }	
        }	
    }

**数据源读取**

数据源创建初始数据集，例如来自文件或Java集合。创建数据集的一般机制是在InputFormat后面抽象的 。Flink附带了几种内置格式，可以从通用文件格式创建数据集。其中许多都在ExecutionEnvironment上有快捷方法。

**基于文件的：**

*  readTextFile(path)/ TextInputFormat- 按行读取文件并将其作为字符串返回。

*  readTextFileWithValue(path)/ TextValueInputFormat- 按行读取文件并将它们作为StringValues返回。StringValues是可变字符串。

*  readCsvFile(path)/ CsvInputFormat- 解析逗号（或其他字符）分隔字段的文件。返回元组或POJO的DataSet。支持基本的java类型及其Value对应的字段类型。

*  readFileOfPrimitives(path, Class)/ PrimitiveInputFormat- 解析新行（或其他字符序列）分隔的原始数据类型（如String或）的文件Integer。

*  readFileOfPrimitives(path, delimiter, Class)/ PrimitiveInputFormat- 解析新行（或其他字符序列）分隔的原始数据类型的文件，例如String或Integer使用给定的分隔符。

**基于集合：**

*  fromCollection(Collection) - 从Java Java.util.Collection创建数据集。集合中的所有元素必须属于同一类型。

*  fromCollection(Iterator, Class) - 从迭代器创建数据集。该类指定迭代器返回的元素的数据类型。

*  fromElements(T ...) - 根据给定的对象序列创建数据集。所有对象必须属于同一类型。

*  fromParallelCollection(SplittableIterator, Class) - 并行地从迭代器创建数据集。该类指定迭代器返回的元素的数据类型。

*  generateSequence(from, to) - 并行生成给定间隔中的数字序列。

**通用：**

*  readFile(inputFormat, path)/ FileInputFormat- 接受文件输入格式。

*  createInput(inputFormat)/ InputFormat- 接受通用输入格式。

**代码示例**

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();	
    	
    // read text file from local files system	
    DataSet<String> localLines = env.readTextFile("file:///path/to/my/textfile");	
    	
    // read text file from a HDFS running at nnHost:nnPort	
    DataSet<String> hdfsLines = env.readTextFile("hdfs://nnHost:nnPort/path/to/my/textfile");	
    	
    // read a CSV file with three fields	
    DataSet<Tuple3<Integer, String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")	
                             .types(Integer.class, String.class, Double.class);	
    	
    // read a CSV file with five fields, taking only two of them	
    DataSet<Tuple2<String, Double>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")	
                                   .includeFields("10010")  // take the first and the fourth field	
                             .types(String.class, Double.class);	
    	
    // read a CSV file with three fields into a POJO (Person.class) with corresponding fields	
    DataSet<Person>> csvInput = env.readCsvFile("hdfs:///the/CSV/file")	
                             .pojoType(Person.class, "name", "age", "zipcode");	
    	
    // read a file from the specified path of type SequenceFileInputFormat	
    DataSet<Tuple2<IntWritable, Text>> tuples =	
     env.createInput(HadoopInputs.readSequenceFile(IntWritable.class, Text.class, "hdfs://nnHost:nnPort/path/to/file"));	
    	
    // creates a set from some given elements	
    DataSet<String> value = env.fromElements("Foo", "bar", "foobar", "fubar");	
    	
    // generate a number sequence	
    DataSet<Long> numbers = env.generateSequence(1, 10000000);	
    	
    // Read data from a relational database using the JDBC input format	
    DataSet<Tuple2<String, Integer> dbData =	
        env.createInput(	
          JDBCInputFormat.buildJDBCInputFormat()	
                         .setDrivername("org.apache.derby.jdbc.EmbeddedDriver")	
                         .setDBUrl("jdbc:derby:memory:persons")	
                         .setQuery("select name, age from persons")	
                         .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO))	
                         .finish()	
        );

![640?wx\_fmt=jpeg][640_wx_fmt_jpeg]

[640_wx_fmt_jpeg]: /images/20211122/18958f48add84dd9b2dcb2401127a113.png