如何将 flink 表保存到 csv 文件？答案

【问题标题】：How to save flink table to csv file?如何将 flink 表保存到 csv 文件？
【发布时间】：2018-05-27 17:35:48
【问题描述】：

我正在使用 Flink 1.4.0

我正在尝试将 Table API 查询的结果保存到 CSV 文件，但我收到错误。以下是详细信息：

我的输入文件如下所示：

id、种类、颜色、重量、名称 311,犬,金,75,dog1 312,犬,棕色,22,dog2 313，猫，灰色，8，cat1

我对此运行查询以仅选择犬科动物，我想将其保存到 .csv 文件：

ExecutionEnvironment env = 
ExecutionEnvironment.getExecutionEnvironment(); 
BatchTableEnvironment tableEnv = 
TableEnvironment.getTableEnvironment(env); 

String inputPath = "location-of-source-file"; 
CsvTableSource petsTableSource = CsvTableSource.builder() 
  .path(inputPath) 
  .ignoreFirstLine() 
  .fieldDelimiter(",") 
  .field("id", Types.INT()) 
  .field("species", Types.STRING()) 
  .field("color", Types.STRING()) 
  .field("weight", Types.DOUBLE()) 
  .field("name", Types.STRING()) 
  .build(); 

// Register our table source 
tableEnv.registerTableSource("pets", petsTableSource); 
Table pets = tableEnv.scan("pets"); 

Table counts = pets 
  .groupBy("species") 
  .select("species, species.count as count") 
  .filter("species === 'canine'"); 

// Convert to Dataset and display results
DataSet<Row> result = tableEnv.toDataSet(counts, Row.class); 
result.print(); 

// Write Results to File 
TableSink<Row> sink = new CsvTableSink("/home/hadoop/output/pets.csv", ","); 
counts.writeToSink(sink);

当我运行它时，我看到数据集的结果正在输出：
犬，2

但是我在输出文件中没有得到任何结果，我看到了这些下面的错误。我能做些什么来解决这个问题？谢谢！

2018-05-27 13:29:17,040 INFO [main] typeutils.TypeExtractor (TypeExtractor.java:1873) - 类 org.apache.flink.types.Row 不包含字段字段的 getter 2018-05-27 13:29:17,040 INFO [main] typeutils.TypeExtractor (TypeExtractor.java:1876) - 类 org.apache.flink.types.Row 不包含字段字段的设置器 2018-05-27 13:29:17,040 INFO [main] typeutils.TypeExtractor (TypeExtractor.java:1911) - 类 org.apache.flink.types.Row 不是有效的 POJO 类型，因为并非所有字段都是有效的 POJO 字段。 2018-05-27 13:29:17,047 INFO [main] typeutils.TypeExtractor (TypeExtractor.java:1873) - 类 org.apache.flink.types.Row 不包含字段字段的 getter 2018-05-27 13:29:17,047 INFO [main] typeutils.TypeExtractor (TypeExtractor.java:1876) - 类 org.apache.flink.types.Row 不包含字段字段的设置器 2018-05-27 13:29:17,047 INFO [main] typeutils.TypeExtractor (TypeExtractor.java:1911) - 类 org.apache.flink.types.Row 不是有效的 POJO 类型，因为并非所有字段都是有效的 POJO 字段。

【问题讨论】：

标签： apache-flink

【解决方案1】：

你可能错过了一个

env.execute();

之后

counts.writeToSink(sink);

与立即触发执行的print() 相比，writeToSink() 只是附加一个接收器运算符并要求显式触发执行。

TypeExtractor 的 INFO 消息“只是”告诉您，Row 不能用作 POJO 类型，但在这里没关系。

【讨论】：