我有要求加密解密使用jks文件内Spark自定义项。使用sparkshell运行我的应用程序时，我得到了以下错误

by: java.net.MalformedURLException: no protocol: CDA_KEYSTORE_PT140.jks

我知道，由于spark将udf视为黑匣子，因此其中的任何文件都不会被读取为hdfs文件，因此我尝试使用将文件副本发送到每个执行器的本地工作目录

/usr/hdp/current/spark2-client/bin/spark-shell --files CDA_KEYSTORE_PT140.jks

我的自定义项如下

def impl2(col1:String): String ={
var pilotCrypto=new PilotCryptoImpl
    pilotCrypto.setKey1("sensitive data")
    pilotCrypto.setKey2("sensitive data")
    pilotCrypto.setKey3("sensitive data")
    pilotCrypto.init()
    EncryptionUtil.setCrypto(pilotCrypto)
    val psg = new IvParamSpecGenerator(true)
    val crypto = new JceCryptoImpl
    crypto.setKeystoreURL("CDA_KEYSTORE_PT140.jks")
    crypto.setKeystoreType("JCEKS")

下面是我使用sparkshell命令运行的完整代码。代码在.scala文件中

import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.rank
import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions.broadcast
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{lit, udf}
import org.apache.spark.sql.types.StringType

def createDataframeFromSql(hc: SparkSession, sql: String): DataFrame = {

    var df = hc.sql(sql)
    return df
  }

def impl2(col1:String): String ={

    var pilotCrypto=new PilotCryptoImpl
    pilotCrypto.setKey1("EbT5a8Fuq")
    pilotCrypto.setKey2("aYt2gv6R")
    pilotCrypto.setKey3("9bFp3Gz4k")
    pilotCrypto.init()
    EncryptionUtil.setCrypto(pilotCrypto)
    val psg = new IvParamSpecGenerator(true)
    val crypto = new JceCryptoImpl
    crypto.setKeystoreURL("CDA_KEYSTORE_PT140.jks")
    crypto.setKeystoreType("JCEKS")
    crypto.setKeyAlias(EncryptionUtil.decryptHex("sensitive data"))
    crypto.setKeyPassword(EncryptionUtil.decryptHex("sensitive data"))
    crypto.setCipherTransformation("AES/CBC/PKCS5Padding")
    crypto.setAlgorithmParamSpecGenerator(psg)
    crypto.setEncodeBase64(true)
    crypto.init()
    val string_to_decrypt = col1
    var encryptedBytes1 = col1.getBytes
    var decryptedBytes1 = new String(crypto.decrypt(encryptedBytes1))
    decryptedBytes1
  }

  def processdata(): Unit = {
    try {
      val hc = SparkSession.builder.appName("HivetoSpark").config("spark.sql.warehouse.dir" , "namenode/apps/hive/warehouse").enableHiveSupport().getOrCreate()
      import hc.implicits._
      hc.sql("""set hive.exec.dynamic.partition=true""")
      hc.sql("""set hive.exec.dynamic.partition.mode=nonstrict""")
      hc.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
      hc.sql("""set hive.merge.tezfiles=true""")
      hc.sql("""set hive.merge.smallfiles.avgsize=256000000""")
      hc.sql("""set hive.merge.size.per.task=256000000""")
      hc.sql("""set hive.merge.sparkfiles=true""")
      val start = System.currentTimeMillis()

     val impl2udf = udf(impl2 _)
      var query1 =s"""select * from table"""
      var fraud_trn_df1 = createDataframeFromSql(hc, query1)
      fraud_trn_df1.show(5,false)

      val fraud_trn_df2 = fraud_trn_df1.withColumn("FT_PRIM_NUM_AMT_decr",impl2udf(col("ft_prim_num_amt")))
      val fraud_trn_df3 = fraud_trn_df2.withColumn("FT_SECONDARY_NUMBER_AMOUNT_decr",impl2udf(col("ft_secndy_num_amt")))
      fraud_trn_df3.show(5,false)

      val end = System.currentTimeMillis()
      val runTimeInSec = (end - start) / 1000.0
      println(s"runTimeInSec: ${runTimeInSec}sec")
    }

最后我调用processdata。

stackrace is as below

Caused by: org.apache.spark.SparkException: Failed to execute user defined function($anonfun$1: (string) => string)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply_133$(Unknown Source)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
        at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:232)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:225)
        at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
        at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:827)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
        at org.apache.spark.scheduler.Task.run(Task.scala:99)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:322)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.net.MalformedURLException: no protocol: CDA_KEYSTORE_PT140.jks
        at java.net.URL.<init>(URL.java:593)
        at java.net.URL.<init>(URL.java:490)
        at java.net.URL.<init>(URL.java:439)
        at com.telus.framework.crypto.impl.jce.JceCryptoImpl.init(JceCryptoImpl.java:78)
        at $line28.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.impl2(<console>:56)
        at $line29.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(<console>:59)
        at $line29.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$1.apply(<console>:59)