Производительность подключенных компонентов GraphFrames

Question

Когда я пытаюсь сгенерировать связанные компоненты с помощью графических фреймов, это занимает значительно больше времени, чем я ожидал. Я работаю на спарке 2.1, графических фреймах 0.5 и AWS EMR с 3 экземплярами r4.xlarge. При генерации подключенных компонентов для графа около 12 миллионов ребер это занимает около 3 часов.

Код ниже. Я довольно новичок, чтобы зажечь, поэтому любые предложения будут потрясающими.

def main(args: Array[String]): Unit = {
  val sparkConf = new SparkConf()
    .setMaster("yarn-cluster")
    .setAppName("Connected Component")

  val sc = new SparkContext(sparkConf)
  sc.setCheckpointDir("s3a://......")
  AWSUtils.setS3Credentials(sc.hadoopConfiguration)

  implicit val sqlContext = SQLContext.getOrCreate(sc)
  import sqlContext.implicits._

  val historical = sqlContext
    .read
    .option("mergeSchema", "false")
    .parquet("s3a://.....")
    .map(x => (x(0).toString, x(2).toString, x(1).toString, x(3).toString, x(4).toString.toLong, x(5).toString.toLong))

  // Complete graph
  val g = GraphFrame(
    historical.flatMap(e => List((e._1, e._3, e._5), (e._2, e._4, e._5))).toDF("id", "type", "timestamp"),
    historical.toDF("src", "dst", "srcType", "dstType", "timestamp", "companyId")
  )

  val connectedComponents: DataFrame = g.connectedComponents.run()

  connectedComponents.toDF().show(100, false)

  sc.stop()
}