我需要使用 Spark Streaming 將某些元素的加密版本添加到復雜的嵌套結構中。進來的 JSON 元素可以有不同的模式，因此我正在尋找一個動態解決方案，我不需要在上面對 Spark 模式進行硬編碼。
例如，這是我可以獲得的 JSON 之一：
{
"hello":"world",
"thisisastruct":
{
"thisisanarray":
[
"thisisanotherstruct":
{
"ID":1,
"thisisthevaluetoenrcypt": "imthevaluetoencrypt"
}
]
}
}
我想完成的是獲得以下內容：
{
"hello":"world",
"thisisastruct":
{
"thisisanarray":
[
"thisisanotherstruct":
{
"ID":1,
"thisisthevaluetoenrcypt": "imthevaluetoencrypt",
"thisisthevaluetoenrcypt_masked": "BNHFBYHTYBFDBY"
}
]
}
}
就像我提到的，模式可能不同，所以我也可能會得到這樣的東西：
{
"hello":"world",
"thisisastruct":
{
"thisisanarray":
[
"thisisanotherstruct":
{
"onemorestruct":
{
"ID":1,
"thisisthevaluetoenrcypt": "imthevaluetoencrypt"
}
}
],
"thisisanothervaluetoenrcypt": "imtheothervaluetoencrypt"
}
}
我想得到這樣的東西：
{
"hello":"world",
"thisisastruct":
{
"thisisanarray":
[
"thisisanotherstruct":
{
"onemorestruct":
{
"ID":1,
"thisisthevaluetoencrypt": "imthevaluetoencrypt",
"thisisthevaluetoencrypt_masked": "BNHFBYHTYBFDBY"
}
}
],
"thisisanothervaluetoencrypt": "imtheothervaluetoencrypt",
"thisisanothervaluetoencrypt_masked": "TYHRBVTRHTYJTJ"
}
}
我有一個 python 方法來加密值；但是，我無法動態更改結構。我認為這樣的事情可能會有所幫助，但不幸的是我沒有 Scala 經驗，我無法將其轉換為 pyspark，并對其進行更改以添加一個新欄位而不是更改當前值
更改 DataFrame 中嵌套列的值
任何幫助將不勝感激
編輯：這是我用來加密資料的功能。我是通過 UDF 進行的，但可以根據需要進行更改
def encrypt_string(s):
result = []
kms = boto3.client('kms', region_name = 'us-west-2')
response = kms.encrypt(
KeyId=key_id,
Plaintext= str(s)
)
return response['CiphertextBlob']
uj5u.com熱心網友回復：
由于您JSON可以使用截然不同的模式將它們決議為 Spark 結構，因此您將需要一個模式，該模式將是所有可能模式的聯合，這很笨拙，因為您不知道它們JSON會是什么樣子。因此，我建議保持JSON字串原樣，并使用 anUDF將字串決議為dict并更新值并將結果作為JSON字串回傳。
from typing import Dict, Any, List
from pyspark.sql import functions as F
from pyspark.sql.types import StringType
import json
data = [('{"hello": "world", "thisisastruct": {"thisisanarray": [{"thisisanotherstruct": {"ID": 1, "thisisthevaluetoenrcypt": "imthevaluetoencrypt"}}]}}', ),
('{"hello": "world", "thisisastruct": {"thisisanarray": [{"thisisanotherstruct": {"onemorestruct": {"ID": 1, "thisisthevaluetoenrcypt": "imthevaluetoencrypt"}}}], "thisisanothervaluetoenrcypt": "imtheothervaluetoencrypt"}}', ), ]
df = spark.createDataFrame(data, ("json_col", ))
def encrypt(s: str) -> str:
return f"encrypted_{s}"
def walk_dict(struct: Dict[str, Any], fields_to_encrypt: List[str]):
keys_copy = set(struct.keys())
for k in keys_copy:
if k in fields_to_encrypt and isinstance(struct[k], str):
struct[f"{k}_masked"] = encrypt(struct[k])
else:
walk_fields(struct[k], fields_to_encrypt)
def walk_fields(field: Any, fields_to_encrypt: List[str]):
if isinstance(field, dict):
walk_dict(field, fields_to_encrypt)
if isinstance(field, list):
[walk_fields(e, fields_to_encrypt) for e in field]
def encrypt_fields(json_string: str) -> str:
fields_to_encrypt = ["thisisthevaluetoenrcypt", "thisisanothervaluetoenrcypt"]
as_json = json.loads(json_string)
walk_fields(as_json, fields_to_encrypt)
return json.dumps(as_json)
field_encryption_udf = F.udf(encrypt_fields, StringType())
df.withColumn("encrypted", field_encryption_udf(F.col("json_col"))).show(truncate=False)
輸出
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|json_col |encrypted |
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|{"hello": "world", "thisisastruct": {"thisisanarray": [{"thisisanotherstruct": {"ID": 1, "thisisthevaluetoenrcypt": "imthevaluetoencrypt"}}]}} |{"hello": "world", "thisisastruct": {"thisisanarray": [{"thisisanotherstruct": {"ID": 1, "thisisthevaluetoenrcypt": "imthevaluetoencrypt", "thisisthevaluetoenrcypt_masked": "encrypted_imthevaluetoencrypt"}}]}} |
|{"hello": "world", "thisisastruct": {"thisisanarray": [{"thisisanotherstruct": {"onemorestruct": {"ID": 1, "thisisthevaluetoenrcypt": "imthevaluetoencrypt"}}}], "thisisanothervaluetoenrcypt": "imtheothervaluetoencrypt"}}|{"hello": "world", "thisisastruct": {"thisisanarray": [{"thisisanotherstruct": {"onemorestruct": {"ID": 1, "thisisthevaluetoenrcypt": "imthevaluetoencrypt", "thisisthevaluetoenrcypt_masked": "encrypted_imthevaluetoencrypt"}}}], "thisisanothervaluetoenrcypt": "imtheothervaluetoencrypt", "thisisanothervaluetoenrcypt_masked": "encrypted_imtheothervaluetoencrypt"}}|
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/378888.html