pyspark生成特定列的行哈希并将其添加为新列

Question

我正在使用 Spark 2.2.0 和 pyspark2。

我已经创建了一个 DataFrame df，现在尝试添加一个新列 “rowhash”，它是 DataFrame 中特定列的 sha2 哈希值。

例如，假设 df 具有以下列:(column1、column2、...、column10)

我需要在新列“rowhash”中使用sha2((column2||column3||column4||……column8), 256)。

目前，我尝试使用以下方法:

1)使用了hash()函数，但由于它给出了整数输出，所以没有多大用处

2) 尝试使用 sha2() 函数，但失败。

假设columnarray有我需要的列数组。

def concat(columnarray):
    concat_str = ''
    for val in columnarray:
        concat_str = concat_str + '||' + str(val) 
    concat_str = concat_str[2:] 
    return concat_str 

然后

df1 = df1.withColumn("row_sha2", sha2(concat(columnarray),256))

此操作因“无法解析”错误而失败。

谢谢你的回答。由于我必须仅对特定列进行哈希处理，因此我创建了这些列名称的列表(在 hash_col 中)并将您的函数更改为:

 def sha_concat(row, columnarray):
   row_dict = row.asDict()      #transform row to a dict
   concat_str = '' 
   for v in columnarray: 
       concat_str = concat_str + '||' + str(row_dict.get(v)) 
   concat_str = concat_str[2:] 
   #preserve concatenated value for testing (this can be removed later)
   row_dict["sha_values"] = concat_str  
   row_dict["sha_hash"] = hashlib.sha256(concat_str).hexdigest()
   return Row(**row_dict) 

然后传递为:

    df1.rdd.map(lambda row: sha_concat(row,hash_col)).toDF().show(truncate=False)

但是现在失败并出现错误:

    UnicodeEncodeError: 'ascii' codec can't encode character u'\ufffd' in position 8: ordinal not in range(128)

我可以在其中一列中看到\ufffd 的值，所以我不确定是否有办法处理这个问题？

Answer 1

您可以使用pyspark.sql.functions.concat_ws()连接您的列和 pyspark.sql.functions.sha2()获取 SHA256 哈希值。

使用来自@gaw的数据:

from pyspark.sql.functions import sha2, concat_ws
df = spark.createDataFrame(
    [(1,"2",5,1),(3,"4",7,8)],
    ("col1","col2","col3","col4")
)
df.withColumn("row_sha2", sha2(concat_ws("||", *df.columns), 256)).show(truncate=False)
#+----+----+----+----+----------------------------------------------------------------+
#|col1|col2|col3|col4|row_sha2                                                        |
#+----+----+----+----+----------------------------------------------------------------+
#|1   |2   |5   |1   |1b0ae4beb8ce031cf585e9bb79df7d32c3b93c8c73c27d8f2c2ddc2de9c8edcd|
#|3   |4   |7   |8   |57f057bdc4178b69b1b6ab9d78eabee47133790cba8cf503ac1658fa7a496db1|
#+----+----+----+----+----------------------------------------------------------------+

您可以传入 0 或 256 作为 sha2() 的第二个参数，根据文档:

Returns the hex string result of SHA-2 family of hash functions (SHA-224, SHA-256, SHA-384, and SHA-512). The numBits indicates the desired bit length of the result, which must have a value of 224, 256, 384, 512, or 0 (which is equivalent to 256).

函数concat_ws接受一个分隔符和要连接的列的列表。我传入 || 作为分隔符，并传入 df.columns 作为列列表。

我在这里使用了所有列，但您可以指定您想要的任何列子集 - 在您的情况下，这将是 columnarray。 (您需要使用 * 来解压列表。)