How to get element by Index in Spark RDD (Java)

by

This should be possible by first indexing the RDD. The transformation zipWithIndex provides a stable indexing, numbering each element in its original order.

Given: rdd = (a,b,c)

val withIndex = rdd.zipWithIndex // ((a,0),(b,1),(c,2))

To lookup an element by index, this form is not useful. First we need to use the index as key:

val indexKey = withIndex.map{case (k,v) => (v,k)}  //((0,a),(1,b),(2,c))

Now, it’s possible to use the lookup action in PairRDD to find an element by key:

val b = indexKey.lookup(1) // Array(b)

If you’re expecting to use lookup often on the same RDD, I’d recommend to cache the indexKey RDD to improve performance.

How to do this using the Java API is an exercise left for the reader.

More Related Contents:

  1. Spark read file from S3 using sc.textFile (“s3n://…)
  2. Matrix Multiplication in Apache Spark [closed]
  3. Difference between SparkContext, JavaSparkContext, SQLContext, and SparkSession?
  4. How to handle this scenario in spark? [closed]
  5. Resolving dependency problems in Apache Spark
  6. Add JAR files to a Spark job – spark-submit
  7. How to flatten a struct in a Spark dataframe?
  8. Spark: How to map Python with Scala or Java User Defined Functions?
  9. How to create SparkSession with Hive support (fails with “Hive classes are not found”)?
  10. Spark sql how to explode without losing null values
  11. Pyspark: Exception: Java gateway process exited before sending the driver its port number
  12. java.lang.ClassCastException using lambda expressions in spark job on remote server
  13. Spark Error – Unsupported class file major version
  14. Spark Strutured Streaming automatically converts timestamp to local time
  15. How can I update a broadcast variable in spark streaming?
  16. Apache Spark – foreach Vs foreachPartition When to use What?
  17. spark broadcast variable Map giving null value
  18. PySpark: java.lang.OutofMemoryError: Java heap space
  19. TaskSchedulerImpl: Initial job has not accepted any resources;
  20. java.lang.NoClassDefFoundError: org/apache/spark/Logging
  21. How to set timezone to UTC in Apache Spark?
  22. Is gzip format supported in Spark?
  23. Change output filename prefix for DataFrame.write()
  24. How can I force Spark to execute code?
  25. Retain keys with null values while writing JSON in spark
  26. How to save models from ML Pipeline to S3 or HDFS?
  27. Reading data from Azure Blob with Spark
  28. Converting mysql table to spark dataset is very slow compared to same from csv file
  29. Spark: get number of cluster cores programmatically
  30. Why does SparkSession execute twice for one action?

Leave a Comment