I have a HDFS cluster on 3 nodes, with replication factor 3. So all data is local on each node. On the same nodes, 3 Spark Executors run (the driver runs elsewhere). When I verify the Spark Task Locality level after running a few SQL queries, I would expect to see all Tasks to have NODE_LOCAL locality. However, I see 1/3 of the tasks on NODE_LOCAL and 2/3 on ANY. I have a hard time understanding this.
Specifically, what I want to measure is the locality level of the tasks that actually fetch the HDFS data (in contrast with the locality level of all tasks). Starting from a Spark events file I filter the tasks that are part of a Stage which is part of a spark.sql.execution.
stages = df.filter("Event='SparkListenerJobStart'").filter(col("Properties.`spark.sql.execution.id`").isNotNull())
result = df.filter("Event='SparkListenerTaskEnd'").select(... "Task Info.Locality", "Stage ID", ...).join(stages, "Stage ID")
When I compare the Locality level of the result DataFrame, I see the 1/3 ratio.
When I print the preferred locations of each partition of the RDD, all partitions list all 3 nodes (which is correct as data is replicated on all machines). I notice these prefered locations are listed by their hostname. On the other hand, in the Spark UI executors are listed by IP, so I wonder if this is a reason for confusion on Spark's side when assigning locality labels on Spark jobs.