apache-spark – 在Spark集群上将数据帧写入文件的速度非常慢

2023年11月13日 433次阅读

我有一个将数据帧写入文件的测试程序.通过为每行添加序列号来生成数据帧,例如

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做了一个rdd – numCol分区,每个分区都会导致一个需要额外运行时间的任务.一般而言,分区数量是并行度和额外成本之间的权衡.尝试100个分区,它应该工作得更好.

顺便说一句,官方Guide建议将此数字设置为群集中CPU数量的2到3倍.

点赞