我有一个将数据帧写入文件的测试程序。数据帧是通过为每行添加序列号生成的,例如
1,2,3,4,5,6,7.....11
2,3,4,5,6,7,8.....12
......
数据框中有 100000 行,但我不认为它太大。
当我提交 Spark 任务时,将数据帧写入 HDFS 上的文件需要将近 20 分钟。我想知道为什么它这么慢,以及如何提高性能。
val sc = new SparkContext(conf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val numCol = 11
val arraydataInt = 1 to 100000 toArray
val arraydata = arraydataInt.map(x => x.toDouble)
val slideddata = arraydata.sliding(numCol).toSeq
val rows = arraydata.sliding(numCol).map { x => Row(x: _*) }
val datasetsize = arraydataInt.size
val myrdd = sc.makeRDD(rows.toSeq, arraydata.size - numCol).persist()
val schemaString = "value1 value2 value3 value4 value5 " +
"value6 value7 value8 value9 value10 label"
val schema =
StructType(schemaString.split(" ").map(fieldName => StructField(fieldName, DoubleType, true)))
val df = sqlContext.createDataFrame(myrdd, schema).cache()
val splitsH = df.randomSplit(Array(0.8, 0.1))
val trainsetH = splitsH(0).cache()
val testsetH = splitsH(1).cache()
println("now saving training and test samples into files")
trainsetH.write.save("TrainingSample.parquet")
testsetH.write.save("TestSample.parquet")
最佳答案
转动
val myrdd = sc.makeRDD(rows.toSeq, arraydata.size - numCol).persist()
到
val myrdd = sc.makeRDD(rows.toSeq, 100).persist()
你已经与
arraydata.size - numCol
建立了一个 rdd分区和每个分区将导致需要额外运行时间的任务。一般来说,分区的数量是并行度和额外成本之间的权衡。尝试 100 个分区,它应该工作得更好。BTW,官方Guide建议将此数字设置为集群中 CPU 数量的 2 或 3 倍。
关于apache-spark - 将数据帧写入 Spark 集群上的文件的速度非常慢,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/41371671/