基于现有行在Scala-Spark数据帧中添加新行

Question

我在Json dataframe中有如下数据。

{"nm": 1233, "date": "2017-01-23", "name": [],"id": "9253194"}
{"nm": 1234, "date": "2017-01-23", "name": [],"id": "9253196"}
{"nm": 1235, "date": "2017-01-23", "name": [],"id": "9253195"}

如何在scala中添加带有索引列的新行以插入到elastic search中。

{"create": {"_type": "usd", "_id": "92531964", "_index": "amount"}}
{"nm": 1233, "date": "2017-01-23", "name": [],"id": "9253194"}
{"create": {"_type": "usd", "_id": "92531966", "_index": "amount"}}
{"nm": 1234, "date": "2017-01-23", "name": [],"id": "9253196"}
{"create": {"_type": "usd", "_id": "92531965", "_index": "amount"}}
{"nm": 1235, "date": "2017-01-23", "name": [],"id": "9253195"}

这里的_id是从现有的列派生的，而_type，_index是常量。

Answer 1

使用flatMap

input.flatMap { x => Seq(x, transform(x)) }

由于这些记录具有不同的模式，因此您可能必须将它们输出为字符串。