我是新来的火花。我有一些json 数据作为HttpResponse 提供。我需要将这些数据存储在配置单元表中。每个HttpGet 请求都会返回一个json，它将是表中的一行。因此，我必须将单行作为文件写入配置单元表目录中。
但是我觉得有太多的小文件会降低速度和效率。那么有没有一种方法可以递归地将新行添加到Dataframe 并一次将其写入配置单元表目录。我觉得这也会减少我的 spark 代码的运行时间。
例子：
for(i <- 1 to 10){
newDF = hiveContext.read.json("path")
df = df.union(newDF)
}
df.write()
我了解数据帧是不可变的。有没有办法做到这一点？
任何帮助将不胜感激。谢谢。
【问题讨论】：
标签： scala hadoop apache-spark recursion dataframe
您显然在滥用 Spark。 Apache Spark 是分析系统，而不是数据库 API。像这样使用 Spark 修改 Hive 数据库没有任何好处。它只会带来严重的性能损失，而不会受益于任何 Spark 功能，包括分布式处理。
您应该直接使用 Hive 客户端来执行事务操作。
【讨论】：
如果您可以批量下载所有数据（例如使用 curl 或其他程序的脚本）并首先将其存储在一个文件中（或多个文件，spark 可以一次加载整个目录），那么您可以一次将该文件（或多个文件）全部加载到 spark 中以进行处理。我还会检查 webapi 以将其视为任何端点，以获取您需要的所有数据，而不是一次只获取一条记录。
【讨论】：
curl -w "\n" [url] >> [jsonfile] 这样的东西在所有网址的循环中应该可以工作。您也可以使用 java 或 scala 或您喜欢的任何其他语言进行下载。
您大多是在正确的轨道上，您要做的是获取多个单个记录作为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)
【讨论】：
.map 的选项 jsonArray。我错过了什么吗？
jsonArray的类型是什么？您提到它是一个“json 数组”，所以我认为它是某种数组。