How to Authenticate Kafka Using Kerberos (SASL), Spark, and Jupyter Notebook

July 19th 2021 new story

5

To briefly explain what we are trying to do here: We want to have permission to read and write Kafka topics.

0 reactions

Our Kafka is protected by Kerberos. It means, before we start accessing Kafka, we need to obtain a ticket from Kerberos. To get the ticket we have to provide a keytab — authentication file for each user. All these steps have to be done automatically because when we use commands to access Kafka there won’t be an opportunity to show keytab manually. To get things done we need to specify the right parameters and configurations in the right place.

0 reactions

Here is my environment (your tools and versions may vary but the approach still should work):

0 reactions

• Cloudera Hadoop cluster v. 5+
• Kafka v. 2+ ( topic with Kerberos auth already exists)
• Spark v. 2+
• Kerberos v. 5
• Jupyter Notebook with Pyspark

For the beginning, let’s access the protected Kafka topic with the terminal. The topic access should only be granted if we obtain a ticket from Kerberos for the right user. For this operation, we need to prepare (it will be smoother if all the files will be in the same path): User’s keytab file ( for Kerberos )

0 reactions

For the beginning, let’s access the protected Kafka topic with terminal. Access to the topic should only be granted if we obtain a ticket from Kerberos for the right user. For this operation we need to prepare (it will be smoother if all the files will be on the same path):

0 reactions

• User’s keytab file ( for Kerberos )
• File jaas.conf:

  KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab=”${PATH_TO_YOUR_KEYTAB}“
  principal=”${USER_NAME}@${REALM}”;
  };

• File kafka_security.properties:
  

  security.protocol=SASL_PLAINTEXT

  
  sasl.kerberos.service.name=kafka
  sasl.mechanism=GSSAPI

• File krb5.conf (probably located in /etc/krb5.conf or /etc/kafka/krb5.conf) (see JDK’s Kerberos Requirements for more details)

Then we need to export the variable with

jaas.conf

krb5.conf

0 reactions

export KAFKA_OPTS=” Djava.security.auth.login.config=jaas.conf -Djava.security.krb5.conf=/etc/krb5.conf”

0 reactions

Then we can write and read Kafka topic from Terminal.

0 reactions

For writing:

0 reactions

/bin/kafka-console-producer --broker-list ${KAFKA_BROKERS_WITH_PORTS} --topic ${TOPIC_NAME} --producer.config kafka_security.properties

For reading:

0 reactions

/bin/kafka-console-consumer --bootstrap-server ${KAFKA_BROKERS_WITH_PORTS} --topic ${TOPIC_NAME} --from-beginning --consumer.config kafka_security.properties

Hope everything worked!

0 reactions

Let’s do the same thing using Spark.

0 reactions

The challenge here is that we want Spark to access Kafka not only with the application driver but also with every executor. It means each executor needs to obtain a ticket from Kerberos with our keytab. To make Spark do this, we need to specify the right configurations.

0 reactions

Firstly, we need the same 

jaas.conf

0 reactions

KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab=”${YOUR_KEYTAB_FILE} “
principal=”${USER_NAME}@${REALM}”;
};

Before launching Spark, we also need to export the variable:

0 reactions

export SPARK_KAFKA_VERSION=0.10

In Spark code we will access Kafka with these options (the first 5 is mandatory):

0 reactions

kafka.bootstrap.servers=${KAFKA_BROKERS_WITH_PORTS}
kafka.security.protocol=SASL_PLAINTEXT
kafka.sasl.kerberos.service.name=kafka
kafka.sasl.mechanism=GSSAPI
subscribe=${TOPIC_NAME}
startingOffsets=latest
maxOffsetsPerTrigger=1000

You can pass these options map to:

0 reactions

spark.readStream.
 format("kafka").
 options(myOptionsMap).
 load()

Before starting Spark we can define the shell variable.

0 reactions

JAVA_OPTIONS="-Djava.security.auth.login.config=jaas.conf -Djava.security.krb5.conf=/etc/krb5.conf"

Also, we will need two copies of users Keytab with different names. If we already have one, we can create the second one with the command:

0 reactions

cp $USER_NAME.keytab ${USER_NAME}_2.keytab

And to launch the spark application we should run this command:

0 reactions

spark2-submit \
--master yarn \
--conf "spark.yarn.keytab=${USER_NAME}_2.keytab" \
--conf "spark.yarn.principal=$USER_NAME@$REALM" \
--conf "spark.driver.extraJavaOptions=$JAVA_OPTIONS" \
--conf "spark.executor.extraJavaOptions=$JAVA_OPTIONS" \
--class "org.example.MyClass" \
--jars spark-sql-kafka-0-10_2.11-2.4.0.jar \
--files "jaas.conf","${USER_NAME}.keytab" \
my_spark.jar

Or you can use the same configurations with spark-shell or pyspark.

0 reactions

Note: to allow Spark access HDFS we specify

spark.yarn.keytab

spark.yarn.principal

spark.driver.extraJavaOptions

spark.executor.extraJavaOptions

jaas.conf

${USER_NAME}.keytab

spark-sql-kafka

0 reactions

Hope everything worked!

0 reactions

Let’s do the same trick in PySpark using Jupyter Notebook.

0 reactions

To access the shell environment from python we will use 

os.environ

0 reactions

import os
import sysos.environ[‘SPARK_KAFKA_VERSION’] = ‘0.10’

Then we should configure the Spark session.

0 reactions

spark = SparkSession.builder. \
config(‘spark.yarn.keytab’, ‘${USER_NAME}_2.keytab’).\
config(‘spark.yarn.principal’, ‘$USER_NAME@$REALM’).\
config(‘spark.jars’, ‘spark-sql-kafka-0–10_2.11–2.4.0.jar’).\
config(‘spark.driver.extraJavaOptions’, ‘-Djava.security.auth.login.config=jaas.conf -Djava.security.krb5.conf=/etc/krb5.conf’).\
config(‘spark.executor.extraJavaOptions’, 
‘-Djava.security.auth.login.config=jaas.conf 
-Djava.security.krb5.conf=/etc/krb5.conf’).\
config(‘spark.files’, ‘jaas.conf,${KEYTAB}’).\
.appName(“KafkaSpark”).getOrCreate()

0 reactions

we can connect to Kafka like this:

0 reactions

kafka_raw = spark.readStream. \
format(‘kafka’).\
option(‘kafka.bootstrap.servers’, ${KAFKA_BROKERS_WITH_PORTS}). \
option(‘kafka.security.protocol’,’SASL_PLAINTEXT’). \
option(‘kafka.sasl.kerberos.service.name’,’kafka’). \
option(‘kafka.sasl.mechanism’,’GSSAPI’). \
option(‘startingOffsets’,’earliest’). \
option(‘maxOffestPerTrigger’,10). \
option(‘subscribe’,${TOPIC_NAME}). \
load()

To access the data we can use:

0 reactions

query = kafka_raw. \
    writeStream. \
    format("console"). \
    start()

That’s it. I hope you could find all the configurations you need to access Kafka using Kerberos any way you like.

0 reactions

Also published here.

0 reactions

5