如何将 Flink 数据集元组转换为一列

Question

我有一个图形数据，比如

1 2
1 4
4 1
4 2
4 3
3 2
2 3

但是我找不到将其转换为单列数据集的方法，例如

1
2
1
4
4
1
...

这是我的代码，我使用了scala ListBuffer，但在Flink DataSet中找不到一种方法

val params: ParameterTool = ParameterTool.fromArgs(args)
val env = ExecutionEnvironment.getExecutionEnvironment
env.getConfig.setGlobalJobParameters(params)
val text = env.readTextFile(params.get("input"))
val tupleText = text.map { line =>
val arr = line.split(" ")
(arr(0), arr(1))
}
var x: Seq[(String, String)] = tupleText.collect()
var tempList = new ListBuffer[String]
x.foreach(line => {
tempList += line._1
tempList += line._2
})
tempList.foreach(println)

Answer 1

你可以用flatMap来做到这一点：

// get some input
val input: DataSet[(Int, Int)] = env.fromElements((1, 2), (2, 3), (3, 4))
// emit every tuple element as own record
val output: DataSet[Int] = input.flatMap( (t, out) => {
out.collect(t._1)
out.collect(t._2)
})
// print result
output.print()