Jak uzyskać identyfikator zadania mapy w Sparku?

Question

Czy istnieje sposób na uzyskanie identyfikatora zadania mapy w programie Spark? Na przykład, jeśli każde zadanie mapy wywołuje funkcję zdefiniowaną przez użytkownika, czy mogę pobrać identyfikator tego zadania mapowego z funkcji zdefiniowanej przez użytkownika?

Answer 1

Nie jestem pewien co masz na myśli przez ID mapy zadanie, ale można uzyskać dostęp do informacji o zadaniach stosując 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) 

Podobna funkcjonalność została dodana do PySpark w 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)