简介 ThriftServer是Hive中的HiveServer2的另外一个实现，它底层使用了Spark SQL来处理SQL语句，从而比Hive拥有更高的性能。 ThriftServer是一个JDBC接口，用户可以通过JDBC连接ThriftServer来访问SparkSQL的数据。ThriftServer在启动的时候，会启动一个SparkSQL的应用程序，而通过JDBC连接进来的客户端共同分享这个sparkSQL应用程序的资源，也就是说不同的用户之间可以共享数据。ThriftServer启动时还会开启一个侦听器，等待JDBC客户端的连接和提交查询。所以，在配置ThriftServer的时候，至少要配置ThriftServer的主机名和端口，如果要使用Hive数据的话，还要提供Hive Metastore的URIs。 ThriftServer默认在安装节点上的10000端口起一个JDBC服务，可以通过Beeline或者JDBC客户端代码来连接它，从而执行SQL命令。 如果您需要了解ThriftServer的其他信息，请参见Spark官网：http://spark.apache.org/docs/1.5.1/sql-programming-guide.html#distributed-sql-engine。

操作步骤 一个简单的流处理系统由以下三部分组件组成：数据源 + 接收器 + 处理器。数据源为Kafka，接受器为Streaming中的Kafka数据源接收器，处理器为Streaming。 对Streaming调优，就必须使三个部件的性能都最优化。 数据源调优 在实际的应用场景中，数据源为了保证数据的容错性，会将数据保存在本地磁盘中，而Streaming的计算结果往往全部在内存中完成，数据源很有可能成为流式系统的最大瓶颈点。 对Kafka的性能调优，有以下几个点： 使用Kafka-0.8.2以后版本，可以使用异步模式的新Producer接口。 配置多个Broker的目录，设置多个IO线程，配置Topic合理的Partition个数。 详情请参见Kafka开源文档中的“性能调优”部分：http://kafka.apache.org/documentation.html。 接收器调优 Streaming中已有多种数据源的接收器，例如Kafka、Flume、MQTT、ZeroMQ等，其中Kafka的接收器类型最多，也是最成熟一套接收器。 Kafka包括三种模式的接收器API： KafkaReceiver：直接接收Kafka数据，进程异常后，可能出现数据丢失。 ReliableKafkaReceiver：通过ZooKeeper记录接收数据位移。 DirectKafka：直接通过RDD读取Kafka每个Partition中的数据，数据高可靠。 从实现上来看，DirectKafka的性能会是最好的，实际测试上来看，DirectKafka也确实比其他两个API性能好。因此推荐使用DirectKafka的API实现接收器。 数据接收器作为一个Kafka的消费者，对于它的配置优化，请参见Kafka开源文档：http://kafka.apache.org/documentation.html。 处理器调优 Streaming的底层由Spark执行，因此大部分对于Spark的调优措施，都可以应用在Streaming之中，例如： 数据序列化 配置内存 设置并行度 使用External Shuffle Service提升性能 在做Spark Streaming的性能优化时需注意一点，越追求性能上的优化，Streaming整体的可靠性会越差。例如： “spark.streaming.receiver.writeAheadLog.enable”配置为“false”的时候，会明显减少磁盘的操作，提高性能，但由于缺少WAL机制，会出现异常恢复时，数据丢失。 因此，在调优Streaming的时候，这些保证数据可靠性的配置项，在生产环境中是不能关闭的。

Spark应用开发流程介绍 Spark包含Spark Core、Spark SQL和Spark Streaming三个组件，其应用开发流程相同。 开发流程中各阶段的说明如图1和表1所示。 图1 Spark应用程序开发流程 表1 Spark应用开发的流程说明 阶段 说明 参考文档 了解基本概念 在开始开发应用前，需要了解Spark的基本概念，根据实际场景选择需要了解的概念，分为Spark Core基本概念、Spark SQL基本概念和Spark Streaming基本概念。 Spark应用开发常用概念 准备开发环境 Spark的应用程序支持使用Scala、Java、Python三种语言进行开发。推荐使用IDEA工具，请根据指导完成不同语言的开发环境配置。 请参考准备Spark应用Java开发环境至准备Spark应用Python开发环境章节 准备运行环境 Spark的运行环境即Spark客户端，请根据指导完成客户端的安装和配置。 准备Spark应用运行环境 获取并导入样例工程 或者新建工程 Spark提供了不同场景下的样例程序，您可以导入样例工程进行程序学习。或者您可以根据指导，新建一个Spark工程。 导入并配置Spark样例工程 根据场景开发工程 提供了Scala、Java、Python三种不同语言的样例工程，还提供了Streaming、SQL、JDBC客户端程序以及Spark on HBase四种不同场景的样例工程。帮助用户快速了解Spark各部件的编程接口。 请参考场景说明至Scala样例代码章节 编译并运行程序 指导用户将开发好的程序编译并提交运行。 编包并运行Spark应用 查看程序运行结果 程序运行结果会写在用户指定的路径下。用户还可以通过UI查看应用运行情况。 查看Spark应用调测结果 调优程序 您可以根据程序运行情况，对程序进行调优，使其性能满足业务场景诉求。 调优完成后，请重新进行编译和运行 请参考数据序列化至Spark CBO调优章节 父主题： Spark应用开发概述

问题 执行Spark Core应用，尝试收集大量数据到Driver端，当Driver端内存不足时，应用挂起不退出，日志内容如下。 16/04/19 15:56:22 ERROR Utils: Uncaught exception in thread task-result-getter-2 java.lang.OutOfMemoryError: Java heap space at java.lang.reflect.Array.newArray(Native Method) at java.lang.reflect.Array.newInstance(Array.java:75) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1671) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1707) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371) at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:71) at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:91) at org.apache.spark.scheduler.DirectTaskResult.value(TaskResult.scala:94) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply$mcV$sp(TaskResultGetter.scala:66) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1716) at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(TaskResultGetter.scala:56) 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) Exception in thread "task-result-getter-2" java.lang.OutOfMemoryError: Java heap space at java.lang.reflect.Array.newArray(Native Method) at java.lang.reflect.Array.newInstance(Array.java:75) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1671) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2000) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1924) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1801) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1351) at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1707) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1345) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:371) at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:71) at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:91) at org.apache.spark.scheduler.DirectTaskResult.value(TaskResult.scala:94) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply$mcV$sp(TaskResultGetter.scala:66) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.scheduler.TaskResultGetter$$anon$3$$anonfun$run$1.apply(TaskResultGetter.scala:57) at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1716) at org.apache.spark.scheduler.TaskResultGetter$$anon$3.run(TaskResultGetter.scala:56) 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)

回答 用户尝试收集大量数据到Driver端，如果Driver端的内存不足以存放这些数据，那么就会抛出OOM(OutOfMemory)的异常，然后Driver端一直在进行GC，尝试回收垃圾来存放返回的数据，导致应用长时间挂起。 解决措施： 如果用户需要在OOM场景下强制将应用退出，那么可以在启动Spark Core应用时，在客户端配置文件“$SPARK_HOME/conf/spark-defaults.conf”中的配置项“spark.driver.extraJavaOptions”中添加如下内容： -XX:OnOutOfMemoryError='kill -9 %p'

操作步骤 打开IDEA工具，选择“Create New Project”。 图1 创建工程 在“New Project”页面，选择“Scala”开发环境，并选择“Scala Module”，然后单击“Next”。如果您需要新建Java语言的工程，选择对应参数即可。 图2 选择开发环境 在工程信息页面，填写工程名称和存放路径，设置JDK版本，并勾选“Config later”（待工程创建完毕后引入scala的编译库文件），然后单击“Finish”完成工程创建。 图3 填写工程信息

环境准备 安装支持环境。（开发环境请参考Spark应用开发环境简介准备） 执行以下命令安装编译工具： yum install cyrus-sasl-devel -y yum install gcc-c++ -y 安装相应的python模块。 需要安装sasl，thrift，thrift-sasl，PyHive。 pip install sasl pip install thrift pip install thrift-sasl pip install PyHive 安装python连接zookeeper工具。 pip install kazoo 从MRS集群上获取相应参数。 zookeeper的IP和PORT： 可以查看配置文件/opt/client/Spark/spark/conf/hive-site.xml中的配置项spark.deploy.zookeeper.url zookeeper 上存放JDBCServer主节点的IP和PORT： 可以查看配置文件/opt/client/Spark/spark/conf/hive-site.xml中的配置项spark.thriftserver.zookeeper.dir（默认是/thriftserver），在此znode子节点（active_thriftserver）上存放了JDBCServer主节点的IP和PORT

样例代码 from kazoo.client import KazooClient zk = KazooClient(hosts='ZookeeperHost') zk.start() result=zk.get("/thriftserver/active_thriftserver") result=result[0].decode('utf-8') JDBCServerHost=result[0].split(":")[0] JDBCServerPort=result[0].split(":")[1] from pyhive import hive conn = hive.Connection(host=JDBCServerHost, port=JDBCServerPort,database='default') cursor=conn.cursor() cursor.execute("select * from test") for result in cursor.fetchall(): print result 其中，ZookeeperHost使用4获取到的zookeeper IP和PORT替换。

回答 Spark部署时，如下jar包存放在客户端的“$SPARK_HOME/jars/streamingClient”目录以及服务端的“/opt/Bigdata/MRS/FusionInsight-Spark-2.2.1/spark/jars/streamingClient”目录： kafka-clients-0.8.2.1.jar kafka_2.10-0.8.2.1.jar spark-streaming-kafka_2.10-1.5.1.jar 由于$SPARK_HOME/lib/streamingClient/*默认没有添加到classpath，所以需要手动配置。 在提交应用程序运行时，在命令中添加如下参数即可： --jars $SPARK_CLIENT_HOME/jars/streamingClient/kafka-clients-0.8.2.1.jar,$SPARK_CLIENT_HOME/jars/streamingClient/kafka_2.10-0.8.2.1.jar,$SPARK_CLIENT_HOME/jars/streamingClient/park-streaming-kafka_2.10-1.5.1.jar 用户自己开发的应用程序以及样例工程都支持上述参数。 但是Spark开源社区提供的KafkaWordCount等样例程序，不仅需要添加--jars参数，还需要配置其他，否则会报“ClassNotFoundException”错误，yarn-client和yarn-cluster模式下稍有不同。 yarn-client模式下 在除--jars参数外，在客户端“spark-defaults.conf”配置文件中，将“spark.driver.extraClassPath”参数值中添加客户端依赖包路径，如“$SPARK_HOME/lib/streamingClient/*”。 yarn-cluster模式下 除--jars参数外，还需要配置其他，有三种方法任选其一即可，具体如下。 在客户端spark-defaults.conf配置文件中，在“spark.yarn.cluster.driver.extraClassPath”参数值中添加服务端的依赖包路径，如“/opt/huawei/Bigdata/FusionInsight/spark/spark/lib/streamingClient/*”。 将各服务端节点的“spark-examples_2.10-1.5.1.jar”包删除。 在客户端“spark-defaults.conf”配置文件中，修改或增加配置选项“spark.driver.userClassPathFirst = true”。

提交SparkLauncher应用程序 在工程目录下执行mvn package命令生成jar包，在工程目录target目录下获取，比如:FemaleInfoCollection.jar 将生成的Jar包（如CollectFemaleInfo.jar）拷贝到Spark运行环境下（即Spark客户端），如“/opt/female”。开启Kerberos认证的安全集群下把从准备Spark应用开发用户中获取的user.keytab和krb5.conf文件拷贝到Spark客户端conf目录下，如：/opt/client/Spark/spark/conf；未开启Kerberos认证集群可不必拷贝user.keytab和krb5.conf文件。 提交SparkLauncher应用程序。 在Spark任务运行过程中禁止重启HDFS服务或者重启所有DataNode实例，否则可能会导致任务失败，并可能导致JobHistory部分数据丢失。 运行程序时可根据需要选择运行模式： --deploy-mode client：driver进程在客户端运行，运行结果在程序运行后直接输出。 --deploy-mode cluster：driver进程在Yarn的ApplicationMaster（AM）中运行，运行结果和日志在Yarn的WebUI界面输出。 java -cp $SPARK_HOME/jars/*:{JAR_PATH} com.huawei.bigdata.spark.examples.SparkLauncherExample yarn-client {TARGET_JAR_PATH} { TARGET_JAR_MAIN_CLASS} {args} JAR_PATH为SparkLauncher应用程序jar包所在路径。 TARGET_JAR_PATH为待提交的spark application应用程序jar包所在路径。 args为待提交的spark application应用程序的参数。

运行“通过JDBC访问Spark SQL”样例程序 在工程目录下执行mvn package命令生成jar包，在工程目录target目录下获取，比如:FemaleInfoCollection.jar 将生成的Jar包（如CollectFemaleInfo.jar）拷贝到Spark运行环境下（即Spark客户端），如“/opt/female”。开启Kerberos认证的安全集群下把从准备Spark应用开发用户中获取的user.keytab和krb5.conf文件拷贝到Spark客户端conf目录下，如：/opt/client/Spark/spark/conf；未开启Kerberos认证集群可不必拷贝user.keytab和krb5.conf文件。 运行“通过JDBC访问Spark SQL”样例程序（Scala和Java语言）。 在Spark任务运行过程中禁止重启HDFS服务或者重启所有DataNode实例，否则可能会导致任务失败，并可能导致JobHistory部分数据丢失。 运行程序时可根据需要选择运行模式： --deploy-mode client：driver进程在客户端运行，运行结果在程序运行后直接输出。 --deploy-mode cluster：driver进程在Yarn的ApplicationMaster（AM）中运行，运行结果和日志在Yarn的WebUI界面输出。 进入Spark客户端目录，使用java -cp命令运行代码。 java -cp ${SPARK_HOME}/jars/*:${SPARK_HOME}/conf:/opt/female/SparkThriftServerJavaExample-*.jar com.huawei.bigdata.spark.examples.ThriftServerQueriesTest ${SPARK_HOME}/conf/hive-site.xml ${SPARK_HOME}/conf/spark-defaults.conf 普通集群需要注释掉安全配置部分代码，详情请参见2和2。 上面的命令行中，您可以根据不同样例工程，最小化选择其对应的运行依赖包。样例工程对应的运行依赖包详情，请参见1。

Spark SQL常用接口 Spark SQL中重要的类有： SQLContext：是Spark SQL功能和DataFrame的主入口。 DataFrame：是一个以命名列方式组织的分布式数据集 DataFrameReader：从外部存储系统加载DataFrame的接口。 DataFrameStatFunctions：实现DataFrame的统计功能。 UserDefinedFunction：用户自定义的函数。 常见的Actions方法有： 表5 Spark SQL方法介绍 方法 说明 Row[] collect() 返回一个数组，包含DataFrame的所有列。 long count() 返回DataFrame的行数。 DataFrame describe(java.lang.String... cols) 计算统计信息，包含计数，平均值，标准差，最小值和最大值。 Row first() 返回第一行。 Row[] head(int n) 返回前n行。 void show() 用表格形式显示DataFrame的前20行。 Row[] take(int n) 返回DataFrame中的前n行。 表6 基本的DataFrame Functions介绍 方法 说明 void explain(boolean extended) 打印出SQL语句的逻辑计划和物理计划。 void printSchema() 打印schema信息到控制台。 registerTempTable 将DataFrame注册为一张临时表，其周期和SQLContext绑定在一起。 DataFrame toDF(java.lang.String... colNames) 返回一个列重命名的DataFrame。 DataFrame sort(java.lang.String sortCol,java.lang.String... sortCols) 根据不同的列，按照升序或者降序排序。 GroupedData rollup(Column... cols) 对当前的DataFrame特定列进行多维度的回滚操作。

操作场景 SQL语句转化为具体执行计划是由SQL查询编译器决定的，同一个SQL语句可以转化成多种物理执行计划，如何指导编译器选择效率最高的执行计划，这就是优化器的主要作用。传统数据库（例如Oracle）的优化器有两种：基于规则的优化器(Rule-Based Optimization,RBO)和基于代价的优化器(Cost-Based Optimization,CBO)。 RBO RBO使用的规则是根据经验形成的，只要按照这个规则去写SQL语句，无论数据表中的内容怎样、数据分布如何，都不会影响到执行计划。 CBO CBO是根据实际数据分布和组织情况，评估每个计划的执行代价，从而选择代价最小的执行计划。 目前Spark的优化器都是基于RBO的，已经有数十条优化规则，例如谓词下推、常量折叠、投影裁剪等，这些规则是有效的，但是它对数据是不敏感的。导致的问题是数据表中数据分布发生变化时，RBO是不感知的，基于RBO生成的执行计划不能确保是最优的。而CBO的重要作用就是能够根据实际数据分布估算出SQL语句，生成一组可能被使用的执行计划中代价最小的执行计划，从而提升性能。 目前CBO主要的优化点是Join算法选择。举个简单例子，当两个表做Join操作，如果其中一张原本很大的表经过Filter操作之后结果集小于BroadCast的阈值，在没有CBO情况下是无法感知大表过滤后变小的情况，采用的是SortMergeJoin算法，涉及到大量Shuffle操作，很耗费性能；在有CBO的情况下是可以感知到结果集的变化，采用的是BroadcastHashJoin算法，会将过滤后的小表BroadCast到每个节点，转变为非Shuffle操作，从而大大提高性能。

操作步骤 Spark CBO的设计思路是，基于表和列的统计信息，对各个操作算子（Operator）产生的中间结果集大小进行估算，最后根据估算的结果来选择最优的执行计划。 设置配置项。 在“spark-defaults.conf”配置文件中增加配置项“spark.sql.cbo”，将其设置为true，默认为false。 在客户端执行SQL语句set spark.sql.cbo=true进行配置。 执行统计信息生成命令，得到统计信息。 此步骤只需在运行所有SQL前执行一次。如果数据集发生了变化（插入、更新或删除），为保证CBO的优化效果，需要对有变化的表或者列再次执行统计信息生成命令重新生成统计信息，以得到最新的数据分布情况。 表：执行COMPUTE STATS FOR TABLE src命令计算表的统计信息，统计信息包括记录条数、文件数和物理存储总大小。 列： 执行COMPUTE STATS FOR TABLE src ON COLUMNS命令计算所有列的统计信息。 执行COMPUTE STATS FOR TABLE src ON COLUMNS name,age命令计算表中name和age两个字段的统计信息。 当前列的统计信息支持四种类型：数值类型、日期类型、时间类型和字符串类型。对于数值类型、日期类型和时间类型，统计信息包括：Max、Min、不同值个数(Number of Distinct Value,NDV)、空值个数(Number of Null)和Histogram（支持等宽、等高直方图）；对于字符串类型，统计信息包括：Max、Min、Max Length、Average Length、不同值个数(Number of Distinct Value,NDV)、空值个数(Number of Null)和Histogram（支持等宽直方图）。 CBO调优 自动优化：用户根据自己的业务场景，输入SQL语句查询，程序会自动去判断输入的SQL语句是否符合优化的场景，从而自动选择Join优化算法。 手动优化：用户可以通过DESC FORMATTED src命令查看统计信息，根据统计信息的分布，人工优化SQL语句。

安全认证代码（Java版） 目前样例代码统一调用LoginUtil类进行安全认证。 在Spark样例工程代码中，不同的样例工程，使用的认证代码不同，基本安全认证或带ZooKeeper认证。样例工程中使用的示例认证参数如表2所示，请根据实际情况修改对应参数值。 表2 参数描述 参数 示例参数值 描述 userPrincipal sparkuser 用户用于认证的账号Principal，您可以联系管理员获取此账号。 userKeytabPath /opt/FIclient/user.keytab 用户用于认证的Keytab文件，您可以联系管理员获取文件。 krb5ConfPath /opt/FIclient/KrbClient/kerberos/var/krb5kdc/krb5.conf krb5.conf文件路径和文件名称。 ZKServerPrincipal zookeeper/hadoop.hadoop.com ZooKeeper服务端principal。请联系管理员获取对应账号。 基本安全认证： Spark Core和Spark SQL程序不需要访问HBase或ZooKeeper，所以使用基本的安全认证代码即可。请在程序中添加如下代码，并根据实际情况设置安全认证相关参数： String userPrincipal = "sparkuser"; String userKeytabPath = "/opt/FIclient/user.keytab"; String krb5ConfPath = "/opt/FIclient/KrbClient/kerberos/var/krb5kdc/krb5.conf"; Configuration hadoopConf = new Configuration(); LoginUtil.login(userPrincipal, userKeytabPath, krb5ConfPath, hadoopConf); 带ZooKeeper认证： 由于“Spark Streaming”、“通过JDBC访问Spark SQL”和“Spark on HBase”样例程序，不仅需要基础安全认证，还需要添加ZooKeeper服务端Principal才能完成安全认证。请在程序中添加如下代码，并根据实际情况设置安全认证相关参数： String userPrincipal = "sparkuser"; String userKeytabPath = "/opt/FIclient/user.keytab"; String krb5ConfPath = "/opt/FIclient/KrbClient/kerberos/var/krb5kdc/krb5.conf"; String ZKServerPrincipal = "zookeeper/hadoop.hadoop.com"; String ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME = "Client"; String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal"; Configuration hadoopConf = new Configuration(); LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, userPrincipal, userKeytabPath); LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZKServerPrincipal); LoginUtil.login(userPrincipal, userKeytabPath, krb5ConfPath, hadoopConf);