csv - 尝试在 Apache Spark 中将 Dataframe 写入 CSV 时出现不一致的行为
问题描述
我正在尝试将使用 Spark 的 MLlib 训练的决策树分类器的最佳超参数输出到使用 Dataframes 和 spark-csv 的 csv 文件。这是我的代码片段:
// Split the data into training and test sets (10% held out for testing)
val Array(trainingData, testData) = assembledData.randomSplit(Array(0.9, 0.1))
// Define cross validation with a hyperparameter grid
val crossval = new CrossValidator()
.setEstimator(classifier)
.setEstimatorParamMaps(paramGrid)
.setEvaluator(new BinaryClassificationEvaluator)
.setNumFolds(10)
// Train model
val model = crossval.fit(trainingData)
// Find best hyperparameter combination and create an RDD
val bestModel = model.bestModel
val hyperparamList = new ListBuffer[(String, String)]()
bestModel.extractParamMap().toSeq.foreach(pair => {
val hyperparam: Tuple2[String,String] = (pair.param.name,pair.value.toString)
hyperparamList += hyperparam
})
val hyperparameters = sqlContext.sparkContext.parallelize(hyperparamList.toSeq)
// Print the best hyperparameters
println(bestModel.extractParamMap().toSeq.foreach(pair => {
println(s"${pair.param.parent} ${pair.param.name}")
println(pair.value)
}))
// Define csv path to output results
var csvPath: String = "/root/results/decision-tree"
val hyperparametersPath: String = csvPath+"/hyperparameters"
val hyperparametersFile: File = new File(hyperparametersPath)
val results = (hyperparameters, hyperparametersPath, hyperparametersFile)
// Convert RDD to Dataframe and write it as csv
val dfToSave = spark.createDataFrame(results._1.map(x => Row(x._1, x._2)))
dfToSave.write.format("csv").mode("overwrite").save(results._2)
// Stop spark session
spark.stop()
完成 Spark 作业后,我可以按预期在路径中看到 part-00*... 和 _SUCCESS 文件。然而,虽然在这种情况下总共有 13 个超参数(通过在屏幕上打印它们来确认),cat
但 csv 文件显示并非每个超参数都写入了 csv:
user@master:~$ cat /root/results/decision-tree/hyperparameters/part*.csv
checkpointInterval,10
featuresCol,features
maxDepth,5
minInstancesPerNode,1
此外,在每次执行中写入的超参数都会发生变化。这是在一个基于 HDFS 的 Spark 集群上执行的,该集群有 1 个 master 和 3 个具有完全相同硬件的 worker。会不会是比赛条件?如果是这样,我该如何解决?
提前致谢。
解决方案
我想我想通了。我希望dfTosave.write.format("csv")save(path)
将所有内容都写入主节点,但由于任务分配给所有工作人员,因此每个工作人员将其部分超参数保存到其文件系统中的本地 CSV。因为在我的例子中,主节点也是一个工作节点,我可以看到它的超参数部分。“不一致的行为”(即在每次执行中看到不同的部分)是由 Spark 用于在工作人员之间分配分区的任何算法引起的。
我的解决方案是使用类似scp
或rsync
构建完整结果的方法从所有工作人员那里收集 CSV。
推荐阅读
- reactjs - 登录后获取数据
- express - 对 DB 的 GET 请求的结果未在服务器日志中定义“错误”
- python - Pandas 条件格式不显示背景颜色。(没有错误)
- django - 仅在所有者的帖子上显示编辑按钮 - Django
- python - 根据 timedelta 将 pandas 帧拆分为多个帧
- flutter - LateError 被抛出构建主页(脏,状态:_HomePageState#5847d):LateInitializationError:字段'imgCamera'尚未初始化
- javascript - 在 Web 组件中设置单个字母样式的方法
- python - 让两个输入之间的欧几里得距离在一个循环内填充矩阵
- node.js - 了解passport.initialize() 和passport.session
- c# - React - 无法在 localhost 中发送 cookie,而在 Postman 和 Swagger 中它可以完美运行