文章目录
- Spark 实现自定义加密
- 一、建立加密和解密的自定义函数
- 二、在 Spark 环境下导入对象实现的方法，并在 SparkSession 中注册 UDF 函数
- 三、在SparkSQL中调用函数
Spark 实现自定义加密
一、建立加密和解密的自定义函数
import java.nio.charset.{StandardCharsets}
import java.util.Base64
import javax.crypto.Cipher
import javax.crypto.spec.SecretKeySpecobject SparkUtil {/*** 处理密钥* @param secret 密钥*/private def secretInit(secret:String)={// 对密钥长度进行约束val allowNumBits: Array[Int] = Array(16, 24, 32)// 如果密钥长度符合，将密钥转换为AES密钥对象if (allowNumBits.contains(secret.size)) {new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8),"AES")}else{throw new RuntimeException(s"AES secret size of numBits ${secret.size} not in permitted values (${allowNumBits.mkString(",")})")}}/*** 加密函数* @param src 源数据* @param secret 密钥*/def encrypt(src:String,secret:String)={// 获取加密算法实例val cipher: Cipher = Cipher.getInstance("AES")// 初始化加密模式，使用给定的密钥(需要先用key()对密钥进行处理)cipher.init(Cipher.ENCRYPT_MODE,secretInit(secret))// 执行加密操作val bytes: Array[Byte] = cipher.doFinal(src.getBytes(StandardCharset.UTF_8))// 返回加密后的数据Base64.getEncoder().encodeToString(bytes)}/*** 解密函数* @param dest 待解密数据* @param secret 密钥*/def decrypt(dest:String,secret:String)={val cipher: Cipher = Cipher.getInstance("AES")cipher.init(Cipher.DECRYPT_MODE,secretInit(secret))val bytes: Array[Byte] = cipher.doFinal(Base64.getDecoder.decode(dest))new String(bytes, StandardCharsets.UTF_8)}
}
二、在 Spark 环境下导入对象实现的方法，并在 SparkSession 中注册 UDF 函数
import core.SparkUtil.{encrypt,decrypt}
spark.udf.register("aes_encrypt",(src:String,secret:String) =>encrypt(src, secret),StringType)
spark.udf.register("aes_decrypt",(src:String,secret:String)=>decrypt(src, secret),StringType)
三、在SparkSQL中调用函数
val frm: DataFrame = spark.createDataFrame(Seq(Test(1,Array("money","freedom"),Map("java"->85,"mysql"->67)),Test(2,Array("beauty","beauty"),Map("java"->72,"mysql"->90)),Test(3,Array("sports","beauty"),Map("java"->76,"html"->52))
))val secret = "henryyb2211ariel"val frmEncrypt: DataFrame = frm.select($"id",callUDF("aes_encrypt",array_join($"hobbies", ","),lit(secret)).as("encrypted_hobbies"))val frmDecrypt: DataFrame = frmEncrypt.select($"id",split(callUDF("aes_decrypt",$"encrypted_hobbies",lit(secret)),",").as("hobbies")).show()
)
