Flink 提供了一套 DataSet 的 API 来做批处理. 其实 DataSet 的使用方法还是和 DataStream 很相似的, 本章主要是先简单的说一下 DataSet 的基本使用.
DataSet API 其实和 DataStream ApI 相似, 都是需要创建 ExecutionEnvironment 环境, 然后通过 ExecutionEnvironment 环境提供的方法读取外部数据, 将外部数据转换为 DataSet 数据集, 之后利用 DataSet 提供的 API 进行转换操作, 并处理成最后的结果, 并对结果进行输出.
DataSources 数据输入
数据输入共有3种类型的接口, 分别是文件系统类型 / Java Collection 类型 / 以及通用数据类型. 其中前两种其实与 DataStream类型, 在前面的系列文章中已经说过了, 这边主要再说一下 通用数据类型接口怎样使用.
DataSet ApI 提供了 Inputformat 通用的数据接口, 已接入不同的数据源和格式类型的数据. Inputformat 接口主要分为两种类型: 一种是基于文件类型, 在 DataSet API 对应的 readFile( ) 方法; 另外一种是基于通用数据类型的接口, 例如读取 RDBMS 或者 NoSQL 数据库等.
下面一个方法就以读取一个csv文件的方式举例, 其中首先定义好了每一行的转换类型, 之后将每一行数据输入都转换为对应的 pojo. 使用 env.createInput() 将 PojoCsvInputFormat 转换为 dataSet.
private static <T> DataSet<T> getSource(ExecutionEnvironment env, String path, String[] fieldOrder, Class<T> type) throws URISyntaxException {
// 本地文件路径
URL fileUrl = InputFormatExample.class.getClassLoader().getResource(path);
Path filePath = Path.fromLocalFile(new File(fileUrl.getPath()));
// 抽取 TypeInformation，是一个 PojoTypeInfo
PojoTypeInfo<T> pojoType = (PojoTypeInfo<T>) TypeExtractor.createTypeInfo(type);
// 由于 Java 反射抽取出的字段顺序是不确定的，需要显式指定下文件中字段的顺序
// 创建 PojoCsvInputFormat
PojoCsvInputFormat<T> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);
return env.createInput(csvInput, pojoType);
}
DataSet 转换操作
Flink 提供了丰富的 API 对 dataSet 做转换处理, 例如数据处理(Map / FlatMap / MapPartiton / Filter), 聚合操作(Reduce / ReduceGroup / Aggregate), 多表关联(Join / OuterJoin / Cogroup / Cross), 集合操作 (Union / Rebalance / Hash-Partition / Range-Partition / Sort Partition), 排序操作(first / minBy / maxBy). 具体的API操作太多, 本文就不一一赘述了,这里就将一些 join 方法的使用.
// 与流处理的合并类型, 也是依靠 where().equalTO()来实现两个dataSet的判断
dataSet1.join(dataSet2).where("key").equalTo("key").with(<JoinFunction>)
下面一个例子展示了如何去使用 join 方法关联两个 dataSet.
// 首先使用上面讲到的方法读取csv文件数据,转为DataSet<POJO>
// item dataSet 格式为(id, price)
String itemPath = "item.csv";
String[] itemField = new String[]{"id", "price"}; // java反射会导致乱序,手动指定字段序
DataSet<Item> items = getSource(env, itemPath, itemField, Item.class);
// info dataSet 格式为(id, color, country)
String infoPath = "info.csv";
String[] infoField = new String[]{"id", "color", "country"};
DataSet<Info> infos = getSource(env, infoPath, infoField, Info.class);
// 关联两个dataset
JoinOperator.DefaultJoin<Item, Info> dataSet = items.join(infos).where("id").equalTo("id");
// 使用 joinFunction 处理合并后的两个dataSet
dataSet.with(new JoinFunction<Item, Info, String>() {
@Override
public String join(Item item, Info info) throws Exception {
return "商品ID:" + item.getId() + " 价格:"+item.getPrice() + " 颜色:"+ info.getColor() + " 国家:" + info.getCountry();
}
}).print();
// 样例数据结果:↓↓↓↓↓
//商品ID:1 价格:50 颜色:red 国家:china
//商品ID:2 价格:120 颜色:black 国家:usa
//商品ID:3 价格:89 颜色:green 国家:korea
DataSinks 数据输出
为了能够让用户更灵活的使用外部数据, Flink抽象出通用的 OutputFormat 接口, 批量数据输出全部实现于此接口.
Flink 内置了常用的数据存储介质对应的接口, 如 TextOutputFormat /CsvOutputFormat / HadoopOutputFormat / JDBCOutputFormat 等.
Flink 在 DataSet ApI 中的数据输出总共以下3类:
1. 基于文件输出接口
WriteAsText / WriteAsCsv. 可以直接使用这两个方法输出到文件, 用户也可以指定写入文件的模式, 分为 OVERWRITE 模式(覆盖) 和 NOT_OVERWRITE 模式(不覆盖 ). 均可以写到hdfs或者本地
2. 通用数据接口
用户可以自己自定义OutputFormat 方法来定义存储, 例如 HadoopOutputFormat.
3. 客户端输出
如果想在本地调试的话, 那么最简单的方式就是通过 print()的方法直接将flink的数据拉回到client, 然后输出.
