如何使用spark结构化流连接到kerberos安全的kafka集群？

hts6caw3 于 2021-06-06 发布在 Kafka

关注(0)|答案(1)|浏览(556)

我正在尝试使用结构化流api连接到kerberos安全的kafka群集。下面是我的代码和spark的输出。我没有看到任何异常，只是警告客户机断开连接的消息。下一步如何解决这个问题？

import org.apache.spark.sql.SparkSession
import org.apache.log4j.{Logger, Level}

object Main {
  def main(args: Array[String]) {
    Logger.getLogger("org").setLevel(Level.WARN)
    Logger.getLogger("akka").setLevel(Level.WARN)

    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("myapp")
      .config("spark.executor.extraJavaOptions", "java.security.auth.login.config=jaas.conf")
      .getOrCreate()

    import spark.implicits._

    val lines = spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "broker1:9100,broker2:9100")
      .option("security.protocol", "SASL_PLAINTEXT")
      .option("sasl.kerberos.service.name", "mysvcname")
        .option("subscribe", "mytopic")
        .load()

    val query = lines.select("value").writeStream.format("console").start()
    query.awaitTermination()
}

输出如下：

Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
19/02/11 17:15:06 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
19/02/11 17:15:10 WARN NetworkClient: [Consumer clientId=consumer-1, groupId=spark-kafka-source-cef02569-ab16-4ca2-a9e8-18bcea992c0d--1359730493-driver-0] Bootstrap broker broker2:9100 (id: -2 rack: null) disconnected
19/02/11 17:15:11 WARN NetworkClient: [Consumer clientId=consumer-1, groupId=spark-kafka-source-cef02569-ab16-4ca2-a9e8-18bcea992c0d--1359730493-driver-0] Bootstrap broker broker1:9100 (id: -1 rack: null) disconnected
19/02/11 17:15:11 WARN NetworkClient: [Consumer clientId=consumer-1, groupId=spark-kafka-source-cef02569-ab16-4ca2-a9e8-18bcea992c0d--1359730493-driver-0] Bootstrap broker broker2:9100 (id: -2 rack: null) disconnected
19/02/11 17:15:11 WARN NetworkClient: [Consumer clientId=consumer-1, groupId=spark-kafka-source-cef02569-ab16-4ca2-a9e8-18bcea992c0d--1359730493-driver-0] Bootstrap broker broker1:9100 (id: -1 rack: null) disconnected
19/02/11 17:15:11 WARN NetworkClient: [Consumer clientId=consumer-1, groupId=spark-kafka-source-cef02569-ab16-4ca2-a9e8-18bcea992c0d--1359730493-driver-0] Bootstrap broker broker1:9100 (id: -1 rack: null) disconnected
19/02/11 17:15:11 WARN NetworkClient: [Consumer clientId=consumer-1, groupId=spark-kafka-source-cef02569-ab16-4ca2-a9e8-18bcea992c0d--1359730493-driver-0] Bootstrap broker broker2:9100 (id: -2 rack: null) disconnected
...

scala apache-kafka apache-spark kerberos

来源：https://stackoverflow.com/questions/54640051/how-do-i-connect-to-a-kerberos-secured-kafka-cluster-with-spark-structured-strea

1条答案

按热度按时间

yhxst69z1#

我解决了我的问题。指定安全协议选项时，选项名称必须以“kafka.”作为前缀。这是令人困惑的，因为对于一个普通的kafka使用者来说，这个选项只是security.protocol，但是为了配置spark，bootstrap.servers和security.protocol（以及您可能需要的任何其他选项/属性）必须以kafka作为前缀。。我最初的代码是：

.option("security.protocol", "SASL_PLAINTEXT")

正确的选择是：

.option("kafka.security.protocol", "SASL_PLAINTEXT")

以下是完整的代码：

import org.apache.spark.sql.SparkSession
import org.apache.log4j.{Level, Logger}

object Main {
  def main(args: Array[String]) {
    Logger.getLogger("org").setLevel(Level.INFO)
    Logger.getLogger("akka").setLevel(Level.INFO)

    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("myapp")
      .config("spark.executor.extraJavaOptions", "java.security.auth.login.config=c:/krb/jaas.conf")
      .getOrCreate()

    import spark.implicits._

    val lines = spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers", "broker1:9100,broker2:9100")
      .option("kafka.security.protocol", "SASL_PLAINTEXT")
      .option("subscribe", "mytopic")
      .load()

    val query = lines.select("value").writeStream.format("console").start()
    query.awaitTermination()
  }
}

以下是jaas.conf文件的内容供参考：

KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="c:/krb/mykeytab.keytab"
  principal="myaccount@mydomain.int"
  storeKey=true
  useTicketCache=false
  serviceName="myservicename";
};

赞(0）回复(0）举报 2021-06-06

我来回答

如何使用spark结构化流连接到kerberos安全的kafka集群？

1条答案

相关问题

热门标签

最新问答