scala - 递归地向数据框添加行

Question

我是 spark 的新手。我有一些作为 HttpResponse 的 json 数据。我需要将这些数据存储在配置单元表中。每个 HttpGet 请求都会返回一个 json，它将成为表中的一行。因此，我不得不将单行作为文件写入配置单元表目录中。

但是我觉得小文件太多会降低速度和效率。那么有没有一种方法可以递归地将新行添加到 Dataframe 并将其一次性写入配置单元表目录。我觉得这也会减少我的 spark 代码的运行时间。

例子:

for(i <- 1 to 10){
 newDF = hiveContext.read.json("path")
 df = df.union(newDF)
}
df.write()

我知道数据帧是不可变的。有办法实现吗？

如有任何帮助，我们将不胜感激。谢谢。

Answer 1

你基本上是在正确的轨道上，你要做的是获取多个单条记录作为一个Seq[DataFrame]，然后减少Seq[DataFrame] 通过合并它们到单个 DataFrame。

根据您提供的代码:

val BatchSize = 100
val HiveTableName = "table"

(0 until BatchSize).
map(_ => hiveContext.read.json("path")).
reduce(_ union _).
write.insertInto(HiveTableName)

或者，如果您想随时执行 HTTP 请求，我们也可以这样做。假设您有一个执行 HTTP 请求并将其转换为 DataFrame 的函数:

def obtainRecord(...): DataFrame = ???

您可以按照以下方式做一些事情:

val HiveTableName = "table"
val OtherHiveTableName = "other_table"
val jsonArray = ???

val batched: DataFrame =
    jsonArray.
    map { parameter =>
      obtainRecord(parameter)
    }.
    reduce(_ union _)
batched.write.insertInto(HiveTableName)
batched.select($"...").write.insertInto(OtherHiveTableName)