Apache Spark - Hive 内连接、LIMIT 和自定义 UDF答案

【问题标题】：Apache Spark - Hive inner join, LIMIT and custom UDFApache Spark - Hive 内连接、LIMIT 和自定义 UDF
【发布时间】：2016-08-24 14:43:40
【问题描述】：

我正在尝试在 hive 中运行查询：

这是最简单的设置（我知道我可以做一个 =，但我使用了一个自定义 UDF，它不仅仅是一个相等比较）

数据集 a 和 b 各有大约 30,000 行

SELECT * FROM a INNER JOIN b ON Custom_UDF_Equals_Comparison(a.id, b.id) LIMIT 5

custom_UDF_Equals_Comparison 只是在 a.id = b.id 之间进行相等性检查

当我运行这个查询时，我可以在我的日志输出中看到很多 m/r 任务正在运行，假设它在两个数据集之间进行比较，直到比较所有可能的排列，并且远高于 5 的 LIMIT（我会预计只有少数 m/r 任务，因为我知道大多数数据可以在每个表的前几行中加入），为什么会发生这种情况？和/或我该如何解决？

编辑：

您好 zero323，这是一个类似的问题，但不准确，它解释了为什么在使用 UDF 进行比较时执行 2 个 RDD 之间的完整比较，但它没有解释为什么 LIMIT 在限制为 5 时不会停止比较成立。例如，如果在前 10 次连接尝试中找到了 5 行，那么为什么剩下的 30,000 * 30,000 次尝试仍然存在。是因为在所有连接发生后才应用限制吗？例如它连接了 30,000*30,000 行，然后将它们减少到只有 5 行？

编辑2：

  def levenshtein(str1: String, str2: String): Int = {
val lenStr1 = str1.length
val lenStr2 = str2.length

val d: Array[Array[Int]] = Array.ofDim(lenStr1 + 1, lenStr2 + 1)

for (i <- 0 to lenStr1) d(i)(0) = i
for (j <- 0 to lenStr2) d(0)(j) = j

for (i <- 1 to lenStr1; j <- 1 to lenStr2) {
  val cost = if (str1(i - 1) == str2(j-1)) 0 else 1

  d(i)(j) = min(
    d(i-1)(j  ) + 1,     // deletion
    d(i  )(j-1) + 1,     // insertion
    d(i-1)(j-1) + cost   // substitution
  )
}

d(lenStr1)(lenStr2)

}

def min(nums: Int*): Int = nums.min

def join_views( joinType: String, parameters: Any, col1: Any, col2: Any) : Boolean = {
if (joinType == "Equals") {
  if (col1 == null || col2 == null) {
    return false
  }

  return col1 == col2
}
else if (joinType == "Fuzzy_String") {
  if (col1 == null || col2 == null) {
    return false
  }

  val val1 = col1.asInstanceOf[String]
  val val2 = col2.asInstanceOf[String]

  val ratio = Utils.distancePercentage(val1, val2)

  if (ratio == 1.0) {
    return val1 == val2
  }

  return (ratio >= parameters.asInstanceOf[Double])
}

return false;

}

... ON join_views("Fuzzy_String", "0.1", a.col1, b.col1) LIMIT 5 = 20secs

... ON join_views("Fuzzy_String", "0.9", a.col1, b.col1) LIMIT 5 = 100secs

【问题讨论】：

它可以保持关闭状态，感谢您的帮助
我实际上发现了一些令人困惑的事情，我的 custom_UDF 也进行了模糊检查，当我以 0.1 必须匹配的模糊值运行它时，它会非常快速地加入并返回结果（例如，它匹配到 5行非常快并返回），当我将其设置为 0.9 时必须匹配它类似于原始 = UDF 中的比较。我想知道为什么 0.1 的模糊匹配返回得更快？不一定是问题，只是观察
这真的很有趣。你能分享一些实施细节吗？
我附上了一些sn-ps，请忽略不良做法:)
当然 :) 你知道o.a.s.sql.functions 提供了开箱即用的levenshtein 吗？

标签： hadoop apache-spark hive hiveql

【解决方案1】：

所以这里有三个不同的问题：

Spark 通过使用散列和排序来优化连接，因此这些优化仅适用于等连接。其他类型的连接，包括依赖于 UDF 的连接，需要成对比较，因此需要笛卡尔积。详情可查看Why using a UDF in a SQL query leads to cartesian product?。
数据移动后的限制操作，尤其是 shuffle，无法完全优化。您可以在Sun Rui 提供的nice answer 到Towards limiting the big RDD 中找到很好的解释。

由于缺少 shuffle，您的情况反而更简单，但您仍然必须将分区放在一起。
限制优化基于分区，而不是记录。 Spark 开始检查第一个分区，如果满足条件的元素数量低于要求，它会在每次迭代中迭代增加分区数量（据我记得因子是 4）。如果您正在寻找罕见的事件，这可能会很快增加。

【讨论】：