How to get ID of a map task in Spark?

by

I am not sure what you mean by ID of map task but you can access task information using TaskContext:

import org.apache.spark.TaskContext

sc.parallelize(Seq[Int](), 4).mapPartitions(_ => {
    val ctx = TaskContext.get
    val stageId = ctx.stageId
    val partId = ctx.partitionId
    val hostname = java.net.InetAddress.getLocalHost().getHostName()
    Iterator(s"Stage: $stageId, Partition: $partId, Host: $hostname")
}).collect.foreach(println)

A similar functionality has been added to PySpark in Spark 2.2.0 (SPARK-18576):

from pyspark import TaskContext
import socket

def task_info(*_):
    ctx = TaskContext()
    return ["Stage: {0}, Partition: {1}, Host: {2}".format(
        ctx.stageId(), ctx.partitionId(), socket.gethostname())]

for x in sc.parallelize([], 4).mapPartitions(task_info).collect():
    print(x)

More Related Contents:

  1. Write to multiple outputs by key Spark – one Spark job
  2. Spark – load CSV file as DataFrame?
  3. Apache Spark: Get number of records per partition
  4. How to Define Custom partitioner for Spark RDDs of equally sized partition where each partition has equal number of elements?
  5. Spark Scala list folders in directory
  6. Spark: long delay between jobs
  7. How to store custom objects in Dataset?
  8. Flattening Rows in Spark
  9. Split 1 column into 3 columns in spark scala
  10. Provide schema while reading csv file as a dataframe
  11. How to flatmap a nested Dataframe in Spark
  12. Joining Spark dataframes on the key
  13. call of distinct and map together throws NPE in spark library
  14. How to get path to the uploaded file
  15. how to filter out a null value from spark dataframe
  16. Is there a reason not to use SparkContext.getOrCreate when writing a spark job?
  17. What are the various join types in Spark?
  18. Replace missing values with mean – Spark Dataframe
  19. How can I force Spark to execute code?
  20. How to calculate the size of dataframe in bytes in Spark?
  21. Why does partition parameter of SparkContext.textFile not take effect?
  22. Exception in thread “main” java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)
  23. Spark-Monotonically increasing id not working as expected in dataframe?
  24. How to remove parentheses around records when saveAsTextFile on RDD[(String, Int)]?
  25. Reading in multiple files compressed in tar.gz archive into Spark [duplicate]
  26. How can I connect to a postgreSQL database into Apache Spark using scala?
  27. Spark Dataframe Random UUID changes after every transformation/action
  28. Spark column string replace when present in other column (row)
  29. How to vectorize DataFrame columns for ML algorithms?
  30. How to set hadoop configuration values from pyspark

Leave a Comment