[SPARK-27232][SQL]Ignore file locality in InMemoryFileIndex if spark.locality.wait is set to zero

[WangGuangxin]

What changes were proposed in this pull request?

InMemoryFileIndex needs to request file block location information in order to do locality schedule in TaskSetManager.

Usually this is a time-cost task. For example, In our production env, there are 24 partitions, with totally 149925 files and 83TB in size. It costs about 10 minutes to request file block locations before submit a spark job. Even though I set spark.sql.sources.parallelPartitionDiscovery.threshold to 24 to make it parallelized, it also needs 2 minutes.

Anyway, this is a waste if we don't care about the locality of files(for example, storage and computation are separate).

So there should be a conf to control whether we need to send getFileBlockLocations request to HDFS NN. If user set spark.locality.wait to 0, file block location information is meaningless.

Here in this PR, if spark.locality.wait is set to 0, it will not request file location information anymore, which will save several seconds to minutes.

How was this patch tested?

tested manually

[AmplabJenkins]

Can one of the admins verify this patch?

[WangGuangxin]

@srowen Could you please help review this patch?

[srowen]

Seems plausible, but I don't really know this code well enough to endorse it. Your best bet is to contact people who may have worked on this code most recently. @peter-toth @adrian-ionescu do you have any thoughts? Maybe @cloud-fan for good measure.

[peter-toth]

Sorry, I don't know this part of the code either. But it looks like PartitionedFileUtil.getBlockLocations can return an empty array even without this change so it looks viable. But I can't judge if it should be bound to spark.locality.wait.

[adrian-ionescu]

I'm not very familiar with this code either, but at a high-level I'd say: let's try to come up with some micro-benchmark that can quantify the perf improvement this brings.

[adrian-ionescu]

I think it'd be safer to check the more specific confs as well and only perform this optimization if they're all 0.

spark/core/src/main/scala/org/apache/spark/internal/config/package.scala

Lines 1160 to 1167 in e4b36df

	private[spark] val LOCALITY_WAIT_PROCESS = ConfigBuilder("spark.locality.wait.process")
	.fallbackConf(LOCALITY_WAIT)
	private[spark] val LOCALITY_WAIT_NODE = ConfigBuilder("spark.locality.wait.node")
	.fallbackConf(LOCALITY_WAIT)
	private[spark] val LOCALITY_WAIT_RACK = ConfigBuilder("spark.locality.wait.rack")
	.fallbackConf(LOCALITY_WAIT)

[peter-toth]

This part doesn't change in this PR. The new thing here is that we don't look up the block locations, but return an empty array instead.

[adrian-ionescu]

Oh, sorry, you're right about that. Please disregard.

[peter-toth]

This part doesn't change in this PR. The new thing here is that we don't look up the block locations, but return an empty array instead.

[squito]

@LantaoJin just pointed me at this based on some discussion in #23951. I totally understand the use case for this, but it needs to use a new config. Even with locality wait == 0, spark still tries to schedule tasks to take advantage of locality. It just means spark won't wait until it gets an offer with better locality. In fact I regularly recommend users to turn locality wait == 0 even on colocated clusters.

Furthermore, even in disagg clusters, you don't necessarily want to turn all locality wait to 0, right? I mean you still might want to wait for locality persisted data from cached rdds?

#23951 pointed out a case for skipping rack resolution entirely on disagg clusters. This is another good case. I'm not entirely sure if they should be controlled by the same thing ... I wonder if there is some hdfs-specific thing which might be appropriate here. Eg. you might have "semi" disagg clusters with most data living remotely, but some small local hdfs. I'm not sure if there is an easy way to figure this out.

[WangGuangxin]

Close this since there is a better solution #24672