Apache Flink：使用 JDBCOutputFormat 时出现异常答案

【问题标题】：Apache Flink: Exception when using JDBCOutputFormatApache Flink：使用 JDBCOutputFormat 时出现异常
【发布时间】：2018-04-13 08:47:43
【问题描述】：

我实现了一个 Flink 程序并尝试将结果 DataSet 存储到关系数据库中。我做了以下命令：

data.output(JDBCOutputFormat.buildJDBCOutputFormat()
                    .setDBUrl(dbURL)
                    .setDrivername(drivername)
                    .setUsername(username)
                    .setPassword(password)
                    .finish()
                    );

dbUrl、drivername、username 和 password 分别存储在 String 中。

我收到以下错误：

The method output(OutputFormat<Tuple8<String,String,String,String,String,String,String,String>>) 
in the type DataSet<Tuple8<String,String,String,String,String,String,String,String>> 
is not applicable for the arguments (JDBCOutputFormat)

我该如何解决这个问题？

【问题讨论】：

请向我们提供完整的例外情况。 DataSet 数据包含什么？ JDBCOutputFormat 只能在 DataSet 包含 Rows（特定的 Flink 类型）时使用。
数据集如下：DataSet<Tuple8<String, String, String, String, String, String, String, String>> data=env.readCsvFile("D:\\WorkDocument\\Dummy_data\\data") .fieldDelimiter("|") .types(String.class, String.class, String.class, String.class, String.class, String.class, String.class, String.class);

标签： apache-flink

【解决方案1】：

从错误消息来看，您的 DataSet 似乎不是 Row 类型（org.apache.flink.types.Row）。将其转换为 Row 类型，然后调用输出函数。

将 Tuple2 类型的变量 'value' 转换为 Row 的示例代码。

Row obj = new Row(2);
obj.setField(0,value.f0);
obj.setField(1,value.f1);

由于您的 DataSet 包含 Tuple8 类型的元素，因此您的 Row 的大小应为 8 (Row(8))。您可以使用映射函数将每个元素从 Tuple8 转换为 Row。

【讨论】：

如何从数据集转换数据集