Spark(Scala)中DataFrame二进制数组列是否有类似startsWith的过滤函数？

结论 #

你给出的写法无法实现需求，原因有两点：

1. Spark原生的startsWith函数仅支持String类型入参，直接对BinaryType的key列传入字节数组调用会触发类型不匹配错误，无法正常执行。
2. 你要匹配的UserId固定占用8字节，你示例中写的5字节前缀长度不符合行键结构，就算支持二进制匹配也无法正确过滤出目标UserId的所有行。

正确实现方案 #

方案1：使用substring函数截取前缀比较（无需自定义UDF） #

Spark的substring函数原生支持BinaryType列，可以直接截取行键前8字节和目标UserId的字节数组做等值比较，代码如下：

import org.apache.spark.sql.functions._
import java.nio.ByteBuffer

// 按照你写入行键时的逻辑，构造目标UserId对应的8字节前缀
val targetUserId = 12345678L
val targetPrefix = ByteBuffer.allocate(8).putLong(java.lang.Long.reverse(targetUserId)).array()

// 截取行键前8字节和目标前缀比较，过滤符合条件的行
val filteredDf = df.filter(substring(col("key"), 1, 8) === targetPrefix)

用你提供的示例数据测试，该逻辑会正确返回第一、第三两条UserId为12345678的记录。

方案2：自定义UDF实现通用前缀匹配 #

如果需要适配可变长度的前缀匹配场景，可以自定义UDF实现字节数组的前缀比较逻辑：

import org.apache.spark.sql.functions._
import java.nio.ByteBuffer

// 定义字节数组前缀匹配UDF
val binaryStartsWith = udf((key: Array[Byte], prefix: Array[Byte]) => {
  if (key == null || prefix == null || key.length < prefix.length) {
    false
  } else {
    prefix.indices.forall(i => key(i) == prefix(i))
  }
})

// 构造目标前缀
val targetUserId = 12345678L
val targetPrefix = ByteBuffer.allocate(8).putLong(java.lang.Long.reverse(targetUserId)).array()

// 调用UDF过滤
val filteredDf = df.filter(binaryStartsWith(col("key"), lit(targetPrefix)))

对接HBase的性能优化建议 #

如果你是直接从HBase读取数据，尽量不要先读全表再在Spark侧过滤，而是将前缀过滤条件下推到HBase侧：
你可以在配置HBase数据源时指定startRow为目标8字节前缀，stopRow为前缀最后一位字节加1的数组，让HBase直接只返回符合前缀的行，避免全表扫描，性能会比Spark侧过滤高几个数量级。

内容的提问来源于stack exchange，提问作者gar.garrison