【spark】spark structrued streaming读写kafka 使用kerberos认证

spark版本:2.4.0
官网
Spark --files使用总结
Spark --files理解

一、编写jar

import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.common.config.SaslConfigs
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.slf4j.{Logger, LoggerFactory}

import scala.collection.mutable

/**
 * 在集群使用以下脚本发送和接受消息！
 * kafka-console-producer.sh --bootstrap-server xxx.01.com:6667,xxx.02.com,xxx.03.com:6667 --topic tp-read  --producer-property security.protocol=SASL_PLAINTEXT
 * kafka-console-consumer.sh --bootstrap-server xxx.01.com:6667,xxx.02.com,xxx.03.com:6667 --topic tp-write --from-beginning  --consumer-property security.protocol=SASL_PLAINTEXT --group tester
 */
object SparkKafka {
   val LOG = LoggerFactory.getLogger(classOf[SparkSession])
  val bootstrapServers = "xxx.01.com:6667,xxx.02.com,xxx.03.com:6667"
  val readTopic = "tp-read"
  val writeTopic = "tp-write"

  def main(args: Array[String]): Unit = {

    val SparkKafkaProps: mutable.Map[String, String] = mutable.Map.empty

    // Kafka’s own configurations can be set via DataStreamReader.option with kafka. prefix, e.g, stream.option("kafka.bootstrap.servers", "host:port").
    // For possible kafka parameters, see Kafka consumer config docs for parameters related to reading data, and Kafka producer config docs for parameters related to writing data.
    SparkKafkaProps.put("kafka." + CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT")
    SparkKafkaProps.put("kafka." + SaslConfigs.SASL_MECHANISM, "GSSAPI")
    SparkKafkaProps.put("kafka." + SaslConfigs.SASL_KERBEROS_SERVICE_NAME, "kafka")
    val spark = SparkSession
      .builder()
      .appName("spark-jar-job")
      .config("spark.sql.streaming.checkpointLocation", "hdfs:///tmp/spark/chkp/test-job")  // 必须设置！
      .enableHiveSupport()
      .getOrCreate()
    // no useful! this api is for spark DStreams not for Structured Streaming!!!
//    val sc=spark.sparkContext
//    val ssc=new StreamingContext(sc,Seconds(5))

    spark.sparkContext.setLogLevel("debug")
    val read = spark
      .readStream
      .format("kafka")
      .options(SparkKafkaProps)
      .option("kafka.bootstrap.servers", bootstrapServers)
      .option("subscribe", readTopic)
      .option("group.id", "sr")
      .option("includeHeaders", "false")
      .load()

       val write=read
          .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
          .writeStream
          .format("kafka")
          .outputMode("update")
          .options(SparkKafkaProps)
          .option("kafka.bootstrap.servers", bootstrapServers)
          .option("topic", writeTopic)
          .trigger(Trigger.ProcessingTime("1 second")) // only change in query
          .start()
    write.awaitTermination()

    LOG.warn("job has started!!!")
    val secConfig = System.getProperty("java.security.auth.login.config")
    LOG.warn(s"Got `java.security.auth.login.config` as: ${secConfig}")

    // You can start any number of queries in a single SparkSession.
    // They will all be running concurrently sharing the cluster resources.
    // You can use sparkSession.streams() to get the StreamingQueryManager
    // block until any one of them terminates 任何一个流结束就会停止！
    // spark.streams.awaitAnyTermination() // 如果定义多流则使用此项。
  }
}

打jar。提交yarn运行

二、启动任务on yarn：

相对路径使用文件名直接表示，绝对路径使用/xxx/文件名表示
spark 任务启动后会有driver和executor。
client模式下的driver就是本机的黑窗口，cluster模式下driver就是就到某一节点机器。
所有模式下executor都是集群的各个节点机器。

# client模式
# 此模式下driver-java-options内的路径就是本机路径，jaas-abs.conf文件内的keytab路径也是本机路径（绝对路径）
# spark.executor.extraJavaOptions 因为是集群的机器，在--files上传到集群各个机器后需要使用相对路径，jaas-rel.conf文件内的keytab路径也是相对路径
# spark.yarn.keytab 文件会和 --files /xxx/user.keytab 文件在上传至集群其他机器后会在同一个文件夹内造成冲突（yarn.Client: Same name resource file:///xxx/user.keytab added multiple times to distributed cache），其实文件是一样的，user-bak.keytab只是user.keytab的一个副本。
# driver-java-options 和 spark.driver.extraJavaOptions 是一样的，但更推荐在client模式下使用driver-java-options
spark-submit \
  --class com.xxx.SparkKafka \
  --verbose \
  --master yarn \
  --deploy-mode client \
  --executor-memory 10G \
  --total-executor-cores 6 \
  --conf spark.yarn.principal=user@XXXXX.COM  \
  --conf spark.yarn.keytab=/xxx/user-bak.keytab \
  --driver-java-options "-Djava.security.auth.login.config=/xxx/jaas-abs.conf" \
  --conf "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=jaas-rel.conf" \
  --files /xxx/jaas-rel.conf,/xxx/user.keytab \
/xxx/spark-kafka-example-0.1.jar

# cluster模式
# 此模式下spark.driver.extraJavaOptions和spark.executor.extraJavaOptions 因为driver运行在是集群的机器，不一定是本机。在--files上传到集群各个机器后需要使用相对路径，jaas-rel.conf文件就是--files上传的，其的keytab路径也是相对路径
# spark.yarn.keytab 文件会和 --files /xxx/user.keytab 文件在上传至集群其他机器后会在同一个文件夹内造成冲突（yarn.Client: Same name resource file:///xxx/user.keytab added multiple times to distributed cache），其实文件是一样的，user-bak.keytab只是user.keytab的一个副本。
spark-submit \
  --class com.xxx.SparkKafka \
  --verbose \
  --master yarn \
  --deploy-mode cluster \
  --executor-memory 10G \
  --total-executor-cores 6 \
  --conf spark.yarn.principal=user@XXXXX.COM  \
  --conf spark.yarn.keytab=/xxx/user-bak.keytab \
  --conf "spark.driver.extraJavaOptions=-Djava.security.auth.login.config=jaas-rel.conf" \
  --conf "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=jaas-rel.conf" \
  --files /xxx/jaas-rel.conf,/xxx/user.keytab \
/xxx/spark-kafka-example-0.1.jar

三、spark shell运行

spark shell其driver运行在本地

bin/spark-shell \
--conf spark.yarn.principal=user@XXXXX.COM  \
--conf spark.yarn.keytab=/xxx/user.keytab \
--driver-java-options "-Djava.security.auth.login.config=/xxx/jaas-abs.conf" \
--conf spark.executor.extraJavaOptions=-Djava.security.auth.login.config=jaas-rel.conf \
--files /xxx/jaas-rel.conf,/xxx/jaas-abs.conf,/xxx/user.keytab \
-Dsun.security.krb5.debug=true \
-Dsun.security.spnego.debug=true \
--verbose

// 测试是否存在user.keytab文件
    val secConfig=System.getProperty("java.security.auth.login.config") // 是指driver端的。
    println(secConfig)
    import java.io.File
    val shortKeyPath=new File("user.keytab")
    shortKeyPath.exists()
    val longKeyPath=new File("/xxx/user.keytab")
    longKeyPath.exists()

// 测试kafka-batch: spark.read
import scala.collection.mutable
val props:mutable.Map[String,String]=mutable.Map.empty
props.put("kafka.bootstrap.servers","xxx.01.com:6667,xxx.02.com,xxx.03.com:6667")
props.put("subscribe","tp-read")
props.put("kafka.security.protocol","SASL_PLAINTEXT")
props.put("kafka.sasl.mechanism","GSSAPI")
props.put("kafka.sasl.kerberos.service.name","kafka")
spark.read.format("kafka").options(props.toMap).load().show



// 测试kafka-stream: spark.readStream
import scala.collection.mutable
import org.apache.spark.sql.streaming.Trigger

val props:mutable.Map[String,String]=mutable.Map.empty
props.put("kafka.bootstrap.servers","xxx.01.com:6667,xxx.02.com,xxx.03.com:6667")
props.put("subscribe","tp-read")
props.put("kafka.security.protocol","SASL_PLAINTEXT")
props.put("kafka.sasl.mechanism","GSSAPI")
props.put("kafka.sasl.kerberos.service.name","kafka")
val rd=spark.readStream.format("kafka").options(props.toMap).load()
rd.writeStream.outputMode("update").format("console").trigger(Trigger.ProcessingTime("3 seconds")).start()  // 可以没有 .awaitAnyTermination()

四、jaas.conf文件内容：

KafkaClient 是kafka客户端使用的，Client 是zookeeper客户端使用的
jaas-abs.conf文件内容

KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  useTicketCache=false
  keyTab="/xxx/user.keytab"
  principal="user@XXXXX.COM";
};

Client {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  keyTab="/xxx/user.keytab"
  principal="user@XXXXX.COM";
}

jaas-rel.conf文件内容
keyTab="user.keytab"和keyTab="./user.keytab"完全一样。

KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  useTicketCache=false
  keyTab="user.keytab"
  principal="user@XXXXX.COM";
};

Client {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  keyTab="user.keytab"
  principal="user@XXXXX.COM";
}

五、其他

如下：报错大多是因为kafka没连接上，也多因为kerberos认证没通过。请仔细检查jaas-*.conf先关的内容。kafka使用jaas.conf配置kerberos。

[Kafka Offset Reader] clients.NetworkClient: [Consumer clientId=consumer-2, groupId=spark-kafka-relation-589f753a-51b2-498d-971e-e513b067ca87-driver-0] Bootstrap broker xxx.com:6667 (id: -3 rack: null) disconnected
2024-10-24 13:17:34,787 WARN  [Kafka Offset Reader] clients.NetworkClient: [Consumer clientId=consumer-2, groupId=spark-kafka-relation-589f753a-51b2-498d-971e-e513b067ca87-driver-0] Bootstrap broker broker xxx.com:6667 (id: -1 rack: null) disconnected