flink使用12-用 table API 实现WordCount

Table API 是 Flink 构建在 DataSet 和 DataStream API 之上的一套结构化编程接口. 本文希望通过一个简单的 wordCount 的例子首先来体验一下普通的 Flink Table 的代码是由哪些部分构成的.

  1. 获取 TableEnvironment

    ExecutionEnvironment 是必不可少的, 不管是Stream API 还是 batch API 都需要一个Environment来管理程序, TableEnvironment 也是在使用 Table API 时首先需要创建的, 它提供了注册内部表/ 执行 Flink SQL 语句/ 注册自定义函数等多种功能. 要获取 TableEnvironment, 首先需要根据情况先创建 DataSet 或者 DataStream 的 Environment, 之后再转换为 TableEnvironment.

    1
    2
    3
    // 以获取 batch Table API 为例, stream类似
    ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);
  2. 拿到 Table

    在以 DataSet API 或者 DataStream API 操作时, 一般有通过集合 / 文件 / socket / 外部数据源 等多种方式来将数据输入. 在 Table API 中, 同样也可以以多种不同的形式去创建一个 Table, 只不过过程相对复杂一些. 拿到一张表的方式有多种, 有通过 Table Descriptor / 用户自定义 Table Source / 由DataStream或者DataSet转换的形式等来实现. 具体的实现操作以后的文章再来详细展示, 下面就以通过 DataSet转换的方式来简单完成.

    1
    2
    3
    4
    5
    6
    7
    // WC 为自定义的 POJO 来统计单词的信息. 首先构建DataSet
    DataSet<WC> input = env.fromElements(
    new WC("Hello", 1),
    new WC("flink", 1),
    new WC("Hello", 1));
    // 直接将 DataSet 转换为 Table
    Table table = tEnv.fromDataSet(input);
  3. 数据转换操作

    拿到了 Table 之后, 就可以正常的进行数据转换的操作了, 如底层的 API一样, 常见的操作 Table API 都已经实现好了, 包括 数据查询和过滤 / 窗口操作 / 聚合操作 / 多表关联 / 集合操作 / 排序操作 / 数据写入. 但是具体的 API语法还是与底层API有很大的不同的, 目的就是为了更方便更类似SQL的形式方便用户去写业务逻辑, 下面就展示一下如何实现 wordCount这个一个步骤.

    1
    2
    3
    4
    5
    Table filtered = table
    .groupBy("word")
    // 在 select 方法中可以方便的直接使用类sql语句进行操作
    .select("word, frequency.sum as frequency")
    .filter("frequency = 2");
  4. 数据输出

    数据处理完毕之后最后一步就是结果的输出, 这一步与底层的API也是大同小异的, 可以将结果直接insetInto()到其他在TableEnvironment注册的表中, 也可以将处理完成的结果转换为 DataSet 或者是 DataStream 亦或是通过自定义的 sink 输出.

    1
    2
    3
    4
    5
    6
    7
    // 1. 直接输入到其他表中
    table.insertInto("otherTbale");
    // 2. 转换为 DataSet / DataStream
    DataSet<WC> result = tEnv.toDataSet(filtered, WC.class);
    // 3. 自定义 sink
    CsvTableSink tableSink = new CsvTableSink(path, ",");
    tEnv.registerTableSink("csvSink", tableSink);

通过上面几步, 一个完整的 Flink Table API 的程序就构建完毕了, 有了前面学习 flink 的经验, 这部分其实是水到渠成的, 稍微了解下就可以上手了, 本文对应的 wordCount 案例在 github 中, 在后面的部分会继续讲解一下其他 Table API 的内容.

0%