ReduceByKey 以字节数组为键

Question

我想使用 Tuple2<byte[], obj> 对的 RDD，但是具有相同内容的 byte[] 被认为是不同的值，因为它们的参考值不同。

我没有看到任何要传入自定义比较器的内容。我可以将 byte[] 转换为具有显式字符集的 String，但我想知道是否有更有效的方法。

Answer 1

您可以创建一个包装器 class 并定义您自己的相等/比较函数。这可能会稍微快一些，因为您不必复制数组（尽管您仍然有一个对象分配）。

Answer 2

自定义比较器不够用，因为 Spark 使用对象的 hashCode 来组织分区中的键。（至少 HashPartitioner 会这样做，你可以提供一个可以处理数组的自定义分区器）

包装数组以提供适当的 equals 和 hashCode 应该可以解决这个问题。轻量级包装器应该可以解决问题：

class SerByteArr(val bytes: Array[Byte]) extends Serializable {
    override val hashCode = bytes.deep.hashCode
    override def equals(obj:Any) = obj.isInstanceOf[SerByteArr] && obj.asInstanceOf[SerByteArr].bytes.deep == this.bytes.deep
}

快速测试：

import scala.util.Random
val data = (1 to 100000).map(_ => Random.nextInt(100).toString.getBytes("UTF-8"))
val rdd = sparkContext.parallelize(data)
val byKey = rdd.keyBy(identity)
// this won't work b/c the partitioner does not support arrays as keys
val grouped = byKey.groupByKey
// org.apache.spark.SparkException: Default partitioner cannot partition array keys.

// let's use the wrapper instead   

val keyable = rdd.map(elem =>  new SerByteArr(elem))
val bySerKey = keyable.keyBy(identity)
val grouped = bySerKey.groupByKey
grouped.count
// res14: Long = 100

ReduceByKey 以字节数组为键

ReduceByKey with a byte array as the key

apache-spark

rdd