zipwithindex 获取 Rdd Key 并获取新的 RDD答案

【问题标题】：zipwithindex for Rdd Key and get new RDDzipwithindex 获取 Rdd Key 并获取新的 RDD
【发布时间】：2019-06-30 08:26:36
【问题描述】：

我正在使用wholeTextfiles 创建rdd。我正在获取文件路径和文件文本。我想要新的RDD，文件路径和索引来自zipWithIndex

我尝试过地图但没有成功。

【问题讨论】：

标签： scala apache-spark rdd

【解决方案1】：

首先需要这样做吗？理论上我们可以做......但是作为要求这样做的必要性是什么。你可以写普通的 hdfs 程序来查找文件名和索引.. 我的意思是 spark rdd 不再需要显示文件名。带索引。

我有以下文件。

现在我正在做如下改造......

import org.apache.log4j.{Level, Logger}
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession

/** *
  * @author : Ram Ghadiyaram
  */
object WholeTextFiles extends Logging {
  Logger.getLogger("org").setLevel(Level.WARN)

  def main(args: Array[String]): Unit = {
    val appName = if (args.length > 0) args(0) else this.getClass.getName
    val spark: SparkSession = SparkSession.builder
      .config("spark.master", "local[*]") //.config("spark.eventLog.enabled", "true")
      .appName(appName)
      .getOrCreate()

// map transformation to form new rdd
    val finalresult = spark.sparkContext.wholeTextFiles("C:\\Users\\Downloads\\codebase\\spark-general-examples\\userdata*.parquet")
    .zipWithIndex().map {
      case (x, index) => (index, x._1)
    }

    println("  print the small rdd this is your tranformed RDD ")

     finalresult.sortByKey(true) foreach {
      case (index,x ) => println(s"\n Index $index file name  ${x}  ")
    }
    println("done")
  }
}

结果：

  print the small rdd this is your tranformed RDD 

 Index 0 file name  file:/C:/Users/Downloads/codebase/spark-general-examples/userdata1.parquet  

 Index 3 file name  file:/C:/Users/Downloads/codebase/spark-general-examples/userdata4.parquet  

 Index 1 file name  file:/C:/Users/Downloads/codebase/spark-general-examples/userdata2.parquet  

 Index 4 file name  file:/C:/Users/Downloads/codebase/spark-general-examples/userdata5.parquet  

 Index 2 file name  file:/C:/Users/Downloads/codebase/spark-general-examples/userdata3.parquet  
done

【讨论】：

我得到了答案。谢谢