PySpark:PicklingError:无法序列化对象:TypeError:无法pickle CompiledFFI 对象

Byr*_*Fox 8 python pickle apache-spark pyspark

我是 PySpark 环境的新手,在尝试使用加密模块加密 RDD 中的数据时遇到错误。这是代码:

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('encrypt').getOrCreate()

df = spark.read.csv('test.csv', inferSchema = True, header = True)
df.show()
df.printSchema()

from cryptography.fernet import Fernet
key = Fernet.generate_key()
f = Fernet(key)

dfRDD = df.rdd
print(dfRDD)
mappedRDD = dfRDD.map(lambda value: (value[0], str(f.encrypt(str.encode(value[1]))), value[2] * 100))
data = mappedRDD.toDF()
data.show()
Run Code Online (Sandbox Code Playgroud)

一切工作,当然很好,直到我试图映射value[1]str(f.encrypt(str.encode(value[1])))。我收到以下错误:

PicklingError:无法序列化对象:TypeError:无法pickle CompiledFFI对象

我还没有看到太多资源提到这个错误,我想看看其他人是否遇到过它(或者你是否通过 PySpark 有推荐的列加密方法)。

hi-*_*zir 4

推荐的列加密方法

您可以考虑 Hive 内置加密(HIVE-5207HIVE-6329),但目前它相当有限(HIVE-7934)。

您当前的代码不起作用,因为Fernet对象不可序列化。您可以通过仅分发密钥来使其工作:

def f(value, key=key): 
    return value[0], str(Fernet(key).encrypt(str.encode(value[1]))), value[2] * 100

mappedRDD = dfRDD.map(f)
Run Code Online (Sandbox Code Playgroud)

或者

def g(values, key=key):
    f = Fernet(key)
    for value in values:
        yield value[0], str(f.encrypt(str.encode(value[1]))), value[2] * 100

mappedRDD = dfRDD.mapPartitions(g)
Run Code Online (Sandbox Code Playgroud)