.NET Core C#中pyspark.sql.functions.hash的等效实现查询

PySpark hash函数核心对齐规则 #

• 底层采用MurmurHash3 32位版本，固定种子为42，这是和通用Murmur3实现差异最大的点
• 输入值的序列化规则完全对齐Spark内部UnsafeRow的序列化逻辑，不是直接对输入转字节进行哈希计算
• 多列输入时按列顺序迭代计算哈希值，最终合并输出结果
• 输入为null时，哈希结果固定返回0

C# 实现代码 #

public static class Murmur3Hash32
{
    private const uint Seed = 42; // 对齐PySpark固定种子
    private const uint C1 = 0xcc9e2d51;
    private const uint C2 = 0x1b873593;

    public static int HashBytes(byte[] data)
    {
        if (data == null || data.Length == 0)
            return 0;

        uint h1 = Seed;
        int length = data.Length;
        int k = 0;
        int index = 0;

        while (length >= 4)
        {
            k = (data[index] & 0xff) |
                ((data[index + 1] & 0xff) << 8) |
                ((data[index + 2] & 0xff) << 16) |
                ((data[index + 3] & 0xff) << 24);

            k *= C1;
            k = RotateLeft(k, 15);
            k *= C2;

            h1 ^= k;
            h1 = RotateLeft(h1, 13);
            h1 = h1 * 5 + 0xe6546b64;

            index += 4;
            length -= 4;
        }

        uint k1 = 0;
        switch (length)
        {
            case 3:
                k1 ^= (uint)(data[index + 2] << 16);
                goto case 2;
            case 2:
                k1 ^= (uint)(data[index + 1] << 8);
                goto case 1;
            case 1:
                k1 ^= data[index];
                k1 *= C1;
                k1 = RotateLeft(k1, 15);
                k1 *= C2;
                h1 ^= k1;
                break;
        }

        h1 ^= (uint)data.Length;
        h1 = FMix(h1);

        return (int)h1;
    }

    // 对齐字符串类型输入，PySpark字符串用UTF-8无BOM编码
    public static int HashString(string input)
    {
        if (input == null)
            return 0;
        var bytes = System.Text.Encoding.UTF8.GetBytes(input);
        return HashBytes(bytes);
    }

    // 对齐Int32类型输入
    public static int HashInt(int input)
    {
        var bytes = BitConverter.GetBytes(input);
        if (BitConverter.IsLittleEndian)
            Array.Reverse(bytes); // Spark内部UnsafeRow整数为大端序
        return HashBytes(bytes);
    }

    private static uint RotateLeft(uint x, int r)
    {
        return (x << r) | (x >> (32 - r));
    }

    private static uint FMix(uint h)
    {
        h ^= h >> 16;
        h *= 0x85ebca6b;
        h ^= h >> 13;
        h *= 0xc2b2ae35;
        h ^= h >> 16;
        return h;
    }
}

验证示例 #

PySpark侧执行代码：

from pyspark.sql import functions as F
spark.sql("select hash('test') as test_hash, hash(123) as int_hash").show()

输出结果参考：

+-----------+----------+
|  test_hash| int_hash|
+-----------+----------+
|-1448094959|1914489594|
+-----------+----------+

C#侧调用验证：

Console.WriteLine(Murmur3Hash32.HashString("test")); // 输出-1448094959
Console.WriteLine(Murmur3Hash32.HashInt(123)); // 输出1914489594

多列哈希计算规则 #

如果需要对齐PySpark多参数的hash(col1, col2, col3)逻辑，按以下步骤实现：

• 按顺序计算第一列的哈希值作为初始值
• 将上一步得到的哈希值作为种子，计算下一列的哈希值
• 依次迭代所有列，最终结果即为多列哈希的输出

常见踩坑提示 #

• 不要直接使用通用Murmur3实现，通用实现默认种子通常为0，和PySpark的固定种子42不匹配
• 处理数值类型时需要对齐Spark的大端序序列化规则，直接用系统默认小端序的字节数组计算会得到错误结果
• 浮点、日期等特殊类型需要先按照Spark内部的序列化规则转换为对应字节数组后再计算哈希

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