/**
   * Create a new cached copy of a DataFrame. For iterative DataFrame-based algorithms.
   *
   * WARNING: This is NOT the same as `DataFrame.cache()`. The original DataFrame will NOT be
   * cached.
   *
   * This is a workaround for SPARK-13346, which makes it difficult to use DataFrames in iterative
   * algorithms. This workaround converts the DataFrame to an RDD, caches the RDD, and creates a
   * new DataFrame. This is important for avoiding the creation of extremely complex DataFrame
   * query plans when using DataFrames in iterative algorithms.
   */
  def getCachedDataFrame(df: DataFrame): DataFrame = {
    val rdd = df.rdd.cache()
    // rdd.count()
    df.sparkSession.createDataFrame(rdd, df.schema)
  }