在 Apache Hadoop 上运行具有本地性质的 Spark 查询时出现数据局部性问题

Posted 2023-04-15

【中文标题】在 Apache Hadoop 上运行具有本地性质的 Spark 查询时出现数据局部性问题

【英文标题】：Problem with data locality when running Spark query with local nature on apache Hadoop

【发布时间】：2021-03-13 11:26:51

【问题描述】：

我有一个 Hadoop 集群，它使用 Apache Spark 来查询保存在 Hadoop 上的 parquet 文件。例如，当我使用以下 PySpark 代码在 parquet 文件中查找单词时：

df = spark.read.parquet("hdfs://test/parquets/*")
df.filter(df['word'] == "jhon").show()

运行此代码后，我转到 spark 应用程序 UI 的阶段选项卡。我看到在Any. 上设置了位置级别摘要。相反，由于此查询的性质，它必须在本地运行，并且至少在NODE_LOCAL 位置级别上运行。当我在运行此程序时检查集群的网络 IO 时，我发现此查询使用网络（查询运行时网络 IO 增加）。这种情况的奇怪之处在于 spark UI 的 shuffle 部分中显示的数字很少。

在 Apache Spark 邮件列表中 Russell Spitzer 的帮助下确定了此问题的根本原因，我运行了以下代码来查找每个分区的首选位置。这段代码的结果让我离解决这个问题又近了一步。我发现首选位置是 IP 形式，也不是主机名，但 spark 使用执行器的 IP 来匹配首选位置并实现数据本地化。

scala> def getRootRdd( rdd:RDD[_] ): RDD[_]  =  if(rdd.dependencies.size == 0) rdd else getRootRdd(rdd.dependencies(0).rdd)
getRootRdd: (rdd: org.apache.spark.rdd.RDD[_])org.apache.spark.rdd.RDD[_]

scala> val rdd = spark.read.parquet("hdfs://test/parquets/*").rdd
rdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[38] at rdd at <console>:24

scala> val scan = getRootRdd(rdd)
scan: org.apache.spark.rdd.RDD[_] = FileScanRDD[33] at rdd at <console>:24

scala> scan.partitions.map(scan.preferredLocations)
res12: Array[Seq[String]] = Array(WrappedArray(datanode-1, datanode-2, datanode-3), WrappedArray(datanode-2, datanode-3, datanode-4), WrappedArray(datanode-3, datanode-4, datanode-5),...

现在我尝试寻找方法让 Spark 首先解析主机名，然后将它们与执行程序的 IP 进行匹配。有什么建议吗？

【参考方案1】：

产生这个问题是因为 Spark 的首选位置 来自 Hadoop 的分区是数据节点主机名， 但 Spark 工作人员通过 IP 注册到 Spark master。 Spark 正在尝试绘制要在 executor 上运行的任务 与本地分区。因为执行器映射到 IP 和分区到主机名，调度程序无法将 IP 与主机名匹配， 并且任务始终在“任何”位置级别上运行。 为了解决这个问题，我们必须运行带有-h [hostname] 标志的spark-workers。 于是，workers通过主机名而不是IP注册到master中，解决了问题。