Spark 3.5中在Spark SQL内使用Hive哈希函数的问题咨询

背景场景 #

现有Spark任务需从Hive 3读取数据并写入MySQL，读取前需执行数据刷新，核心SQL如下：

Insert overwrite schema_x.table_x    
SELECT distinct
         CASE WHEN hash(email_name) < 0 THEN (hash(email_name) * -1)
              WHEN hash(email_name) > 0 THEN hash(email_name)
         END as id,
         trim(email_name) as name,
         camp_id
    from schema.table_name

该任务原通过Airflow的Hive Operator执行，迁移到Spark SQL后，因Spark与Hive的hash函数实现不同，导致结果不一致。针对以下问题逐一解答：

1. 是否可强制Spark使用Hive的hash实现？若可以，如何操作？ #

可以，操作方式如下：

• 初始化SparkSession时启用Hive支持：

  val spark = SparkSession.builder()
    .appName("HiveHashCompatibility")
    .enableHiveSupport()
    .getOrCreate()
  

• 配置参数开启Hive UDF兼容：
  在Spark配置中添加spark.sql.hive.convertLegacyHiveUDFs=true，让Spark兼容Hive的旧版内置函数实现。
• 在SQL中明确调用Hive的hash函数：
  将原SQL中的hash(email_name)改为hive.hash(email_name)，直接调用Hive原生的hash实现。

2. UDF能否解决该问题？是否可在UDF中直接调用Hive的hash而非重实现内部逻辑？ #

UDF可以解决该问题，且无需重写Hive的hash逻辑，可直接调用Hive的原生实现：

• 引入与环境Hive 3匹配的hive-exec依赖包。
• 编写UDF调用Hive的UDFHash类：

  import org.apache.hadoop.hive.ql.udf.UDFHash
  import org.apache.spark.sql.functions.udf
  
  // 定义UDF，调用Hive的hash并取绝对值
  val hiveHashUdf = udf((input: String) => {
    val hiveHash = new UDFHash()
    Math.abs(hiveHash.evaluate(input).asInstanceOf[Int])
  })
  // 注册UDF到SparkSession
  spark.udf.register("hive_hash", hiveHashUdf)
  

• 在SQL中使用该UDF替代原hash函数：hive_hash(email_name)。

3. 是否可通过DataFrame API实现需求？ #

可以，DataFrame API中可集成Hive hash逻辑实现相同需求：

import org.apache.hadoop.hive.ql.udf.UDFHash
import org.apache.spark.sql.functions.{col, trim, udf}

// 定义Hive hash的UDF
val hiveHash = udf((s: String) => {
  val hashVal = new UDFHash().evaluate(s)
  Math.abs(hashVal.asInstanceOf[Int])
})

// 用DataFrame API实现原SQL逻辑
val resultDF = spark.table("schema.table_name")
  .select(
    hiveHash(col("email_name")).alias("id"),
    trim(col("email_name")).alias("name"),
    col("camp_id")
  )
  .distinct()

// 覆盖写入目标表
resultDF.write.mode("overwrite").saveAsTable("schema_x.table_x")

4. 若上述方案不可行，最优解决方式是什么？ #

若因依赖冲突、环境限制等导致上述方案无法实施，最优方式是重实现与Hive完全一致的hash逻辑。
Hive的UDFHash对String类型的处理直接调用Java的String.hashCode()方法，因此只需在Spark中调用该方法并取绝对值，即可得到与Hive完全一致的结果：

• 注册自定义UDF：

  val javaHashUdf = udf((s: String) => Math.abs(s.hashCode()))
  spark.udf.register("hive_compatible_hash", javaHashUdf)
  

• 在SQL或DataFrame中使用该UDF替代原hash函数，即可保证结果一致。

内容的提问来源于stack exchange，提问作者Aishwary Shukla