2. 在resources下创建conf.properties配置文件，内容如下：
```xml
kafka.brokers=cdh1.macro.com:9092,cdh2.macro.com:9092,cdh3.macro.com:9092
kafka.topics=kafka2_spark2
zookeeper.list=cdh1.macro.com:2181
zookeeper.port=2181

1. 编写Spakr程序 ```scala import java.io.{File, FileInputStream} import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord} import org.apache.spark.SparkConf import org.apache.spark.rdd.RDD import org.apache.spark.sql.{SaveMode, SparkSession} import org.apache.spark.streaming.dstream.InputDStream import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies} import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreamingFromKafkaToHive_NotKerberos {

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

// StreamingContext创建
val sparkConf = new SparkConf().setAppName("SparkStreaming")
val ssc = new StreamingContext(sparkConf,Seconds(3))
// SparkSQL
val spark = SparkSession.builder().enableHiveSupport().config(sparkConf).getOrCreate()
// 加载配置文件
val properties = new Properties()
var confPath: String = System.getProperty("user.dir") + File.separator + "conf"+File.separator
val hiveConfPath = new FileInputStream(new File(confPath+"hive_conf.properties"))
if (hiveConfPath == null) {
  println("找不到hive_conf.properties配置文件")
} else {
  properties.load(hiveConfPath)
  println("已加载配置文件")
}
// 读取配置文件
val brokers = properties.getProperty("kafka.brokers")
val topics = properties.getProperty("kafka.topics")
val tableName = properties.getProperty("hive.table.name")
val groupId = properties.getProperty("kafka.group.id")
// kafka配置
val kafkaPara: Map[String, Object] = Map[String, Object](
  ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
  ConsumerConfig.GROUP_ID_CONFIG -> groupId,
  "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
)
val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent, //描述采集的节点跟计算的节点如何匹配
  ConsumerStrategies.Subscribe[String, String](Set(topics), kafkaPara)
)
// 隐式引入
import spark.implicits._
kafkaDataDS.foreachRDD(
  rdd => {
    val unit: RDD[UserInfo] = rdd.map(
      line => {
        val a = line.value()
        val data = a.split(" ")
        val id = data(0)
        val name = data(1)
        val age = data(2)
        new UserInfo(id,name, age)
      }
    )
    val userinfoDF = spark.sqlContext.createDataFrame(unit)
    userinfoDF.write.mode(SaveMode.Append).insertInto(tableName)
  }
)
// 开始运行
ssc.start()
// 等待Streaming运行结束关闭资源
ssc.awaitTermination()

} case class UserInfo(id:String,name:String, age:String)

}

<a name="HTQtb"></a>
#### 3.3.3、程序测试
1. 使用maven工具打jar包后上传至hadoop环境
![image.png](https://cdn.nlark.com/yuque/0/2020/png/2680099/1609137760481-b2e2eb81-8f11-4c79-a2b6-338cf793d931.png#align=left&display=inline&height=114&margin=%5Bobject%20Object%5D&name=image.png&originHeight=114&originWidth=709&size=18034&status=done&style=none&width=709)
2. 使用spark2-submit提交任务
```sql
spark2-submit --class SparkStreamingFromKafka2ToHive2_NotKerberos --master local --jars sparkStreamingFromKafkaToHive-1.0.0.jar

注意：每个节点都需要有如下包

1. java程序模拟kafka生产者发送消息

1. 查询kafka2_spark2_hive2表，消息成功发送

3.5、遇到的问题：

1. 提示找不到视图或表（找不到元数据）

解决方案：
在hadoop环境中拷贝hive-site.xml文件到项目中的resource文件夹下（不知道路径用：find / -name hive-site.xml 搜索）

1. 提示实例化错误，截图如下：

原因：window下需要安装hadoop环境
解决：
1、下载集群对应linux的hadoop版本 <—超链接
2、下载对应的window下的hadoop需要的bin目录 <—超链接
3、解压Linux，把bin目录替换为winutils的bin目录
4、在代码中引用这个地址

System.setProperty("hadoop.home.dir", "E:\\Application\\hadoop-2.6.0")

1. java.lang.IllegalArgumentException: Missing application resource 找不到资源

原因:jar包路径出错
去掉 —jars再次执行成功

3.4、kerberos环境模拟kafka生产者发送消息到队列

3.4.1、程序开发

• 采用keytab形式做安全认证

1. 准备访问Kafka的Keytab文件，使用xst命令导出keytab文件

   kadmin.local 
   xst -norandkey -k userA.keytab userA@MACRO.COM

   

2. 使用klist命令检查导出的keytab文件是否正确

   klist -ek userA.keytab

   

3. 导入keytab文件到java项目的resource文件下

1. 准备producer_jaas.cof文件内容如下：

   KafkaClient{
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="E:\\JavaWorkSpace\\kafka\\conf\\userA.keytab"
   storeKey=true
   useTicketCache=false
   principal="userA@MACRO.COM"
   serviceName=kafka;
   };

   

2. 导入linux的/etc目录下的brk5.conf文件到javaresource目录

1. 准备配置文件，放在项目根目录下的conf目录下

   security.protocol=SASL_PLAINTEXT
   sasl.mechanism=GSSAPI
   sasl.kerberos.service.name=kafka
   bootstrap.servers=cdh1.macro.com:9092
   acks=all
   retries=1
   batch.size=16384
   linger.ms=1
   buffer.memory=33554432
   key.serializer=org.apache.kafka.common.serialization.StringSerializer
   value.serializer=org.apache.kafka.common.serialization.StringSerializer
   topic=kafka2_spark2

2. 编写生产者java代码用作发送消息 ```java import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.*; import java.util.Properties; import java.util.logging.Level; import java.util.logging.Logger;

public class MyProducer {

public static void main(String[] args) {
    Logger.getLogger("org").setLevel(Level.INFO);
    // 获取当前路径
    String localPath = System.getProperty("user.dir") + File.separator;
    String krb5Path = localPath + "conf/krb5.conf";
    String jaasPath = localPath + "conf/jaas.conf";
    if (!new File(krb5Path).exists()) {
        System.out.println("未加载到krb5.conf配置文件，程序结束");
        System.exit(0);
    }
    if (!new File(jaasPath).exists()) {
        System.out.println("未加载到jaas.conf配置文件，程序结束");
        System.exit(0);
    }
    // kerberos必须添加krb5和jaas地址
    //在windows中设置JAAS，也可以通过-D方式传入
    System.setProperty("java.security.krb5.conf", krb5Path);
    System.setProperty("java.security.auth.login.config", jaasPath);
    System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
    // 创建Kafka生产者的配置信息
    Properties properties = new Properties();
    FileInputStream is = null;
    //InputStream ips = ClassLoader.getSystemClassLoader().getResourceAsStream("conf.properties");
    try {
        is = new FileInputStream(new File(localPath+"conf/conf.properties"));
    } catch (FileNotFoundException e) {
        e.printStackTrace();
    }
    try {
        properties.load(is);
    } catch (IOException e) {
        System.out.println("配置文件加载失败！");
    }finally {
        try {
            if (is != null) {
                is.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    // 创建生产者对象
    KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
    // 发送数据
    // 创建流对象
    BufferedReader br = null;
    try {
        br = new BufferedReader(new FileReader("datas/userInfo.txt"));//args[0]  //"datas/userInfo.txt"
        // 定义字符串,保存读取的一行文字
        String line = null;
        // 循环读取,读取到最后返回null
        while ((line = br.readLine()) != null) {
            producer.send(new ProducerRecord<String, String>(properties.getProperty("topic"), line));
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        // 释放资源
        if (br != null) {
            try {
                br.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    // 关闭资源
    producer.close();
}

}

7. 准备datas/userInfo.txt文件，
![image.png](https://cdn.nlark.com/yuque/0/2020/png/2680099/1609308956427-4cf0ea2c-7ef0-4022-94e6-76e8faf0f53a.png#align=left&display=inline&height=248&margin=%5Bobject%20Object%5D&name=image.png&originHeight=325&originWidth=978&size=28988&status=done&style=none&width=746)
8. 测试数据发送成功
![image.png](https://cdn.nlark.com/yuque/0/2020/png/2680099/1609308848373-f66c9b8a-7b5a-4271-9132-b8da107c9ceb.png#align=left&display=inline&height=570&margin=%5Bobject%20Object%5D&name=image.png&originHeight=984&originWidth=1288&size=208914&status=done&style=none&width=746)
9. 消费者端成功接收收据
![image.png](https://cdn.nlark.com/yuque/0/2020/png/2680099/1609308901259-0c1aa6bc-da66-4666-9cb1-a1a13e68f055.png#align=left&display=inline&height=396&margin=%5Bobject%20Object%5D&name=image.png&originHeight=626&originWidth=1178&size=47777&status=done&style=none&width=746)
<a name="ofW12"></a>
## 4.3、打成jar包编写脚本用做发送数据测试
1. 打包之前修改代码，手动指定文件地址
![image.png](https://cdn.nlark.com/yuque/0/2020/png/2680099/1609309133147-56b39fef-cda0-4129-a348-2bb5340d27cc.png#align=left&display=inline&height=210&margin=%5Bobject%20Object%5D&name=image.png&originHeight=218&originWidth=773&size=24222&status=done&style=none&width=746)
2. maven自带工具打包
![image.png](https://cdn.nlark.com/yuque/0/2020/png/2680099/1609309233941-ec3d4401-117b-46a8-9471-b8afd526a527.png#align=left&display=inline&height=282&margin=%5Bobject%20Object%5D&name=image.png&originHeight=563&originWidth=1127&size=80282&status=done&style=none&width=563.5)
3. 上传jar包到linux环境，编写shell脚本，脚本如下
```shell
#!/bin/bash
JAVA_HOME=/usr/java/jdk1.8.0_144/
read_file=$1
for file in `ls lib/*jar`
do
    CLASSPATH=$CLASSPATH:$file
done
export CLASSPATH
${JAVA_HOME}/bin/java -Xms1024m -Xmx2048m MyProducer $read_file

1. 新建目录kafka_kerberos_producer，拷贝idea下conf目录，新建lib目录，移动jar包到lib，目录结构如下

• 运行会报错缺失部分类，移动对应目录到lib目录即可

1. 运行效果如下，test是我们要发送的文件

1. consumer成功接收数据

5、Kerberos环境SparkStreaming拉取kafka数据写入hive

5.1、环境准备

• 采用keytab形式做安全认证

1. 准备访问Kafka的Keytab文件，使用xst命令导出keytab文件

   kadmin.local 
   xst -norandkey -k hive.keytab hive@MACRO.COM

   

2. 使用klist命令检查导出的keytab文件是否正确

   klist -ek hive.keytab

   

3. 准备jaas.cof文件内容如下：

   KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/root/spark/conf/hive.keytab"
   principal="hive@MACRO.COM";
   };
   Client {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   storeKey=true
   keyTab="/root/spark/conf/hive.keytab"
   principal="hive@MACRO.COM";
   };

4. 将hive.keytab和jaas.conf文件拷贝至集群的所有节点统一的/root/spark/conf目录下

1. 准备刚刚编写的Kerberos环境发送数据的脚本

2. 登录CM进入SPARK2服务的配置项将spark_kafka_version的kafka版本修改为0.10

   5.2、程序开发

3. 使用maven创建scala语言的spark2demo工程，pom.xml依赖如下

    <dependencies>
        <!-- Spark依赖关系 -->
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka-0-10 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
      <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
    </dependencies>

4. 在recource下创建conf.properties配置文件，内容如下：

   kafka.brokers=cdh1.macro.com:9092,cdh2.macro.com:9092,cdh3.macro.com:9092
   kafka.topics=kafka2_spark2
   group.id=testgroup

5. 编写spark代码如下： ```scala import java.io.{File, FileInputStream} import java.util.Properties

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord} import org.apache.spark.SparkConf import org.apache.spark.rdd.RDD import org.apache.spark.sql.{SaveMode, SparkSession} import org.apache.spark.streaming.dstream.InputDStream import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies} import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreamingFromKafkaToHive_Kerberos { var confPath: String = System.getProperty(“user.dir”) + File.separator + “conf” + File.separator

def main(args: Array[String]): Unit = { // StreamingContext创建 val sparkConf = new SparkConf().setAppName(“SparkStreaming”)//.setMaster(“local[*]”) val ssc = new StreamingContext(sparkConf, Seconds(3)) // SparkSQL val spark = SparkSession.builder().enableHiveSupport().config(sparkConf).getOrCreate()

// 加载配置文件
val properties = new Properties()
val in = new FileInputStream(new File(confPath + "conf.properties"))
if (in == null) {
  println("找不到conf.properties配置文件")
} else {
  properties.load(in)
  //println("已加载配置文件")
}
// 读取配置文件
val brokers = properties.getProperty("kafka.brokers")
val topics = properties.getProperty("kafka.topics")
val groupId = properties.getProperty("group.id")
val tableName = properties.getProperty("hive.tableName")
// kafka配置
val kafkaPara: Map[String, Object] = Map[String, Object](
  ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
  ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "latest",
  "group.id" -> groupId,
  ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
  ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
  "security.protocol" -> "SASL_PLAINTEXT",
  "sasl.kerberos.service.name" -> "kafka"
)
val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
  ssc,
  LocationStrategies.PreferConsistent, //描述采集的节点跟计算的节点如何匹配
  ConsumerStrategies.Subscribe[String, String](Set(topics), kafkaPara)
)
// 隐式引入
import spark.implicits._
kafkaDataDS.foreachRDD(
  rdd => {
    val unit: RDD[UserInfo] = rdd.map(
      line => {
        val a = line.value()
        val data = a.split(" ")
        val name = data(0)
        val age = data(1)
        new UserInfo(name, age)
      }
    )
    val userinfoDF = spark.sqlContext.createDataFrame(unit)
    userinfoDF.write.mode(SaveMode.Append).insertInto(tableName)
  }
)
// 开始运行
ssc.start()
// 等待Streaming运行结束关闭资源
ssc.awaitTermination()

}

case class UserInfo( name: String, age: String)

}

<a name="lKR9Q"></a>
## 5.3、测试
1. 通过maven自带工具打jar包，上传至linux环境根目录下，运行jar包
```shell
spark2-submit --class SparkStreamingFromKafkaToHive_Kerberos --master local --deploy-mode client --executor-memory 1g --executor-cores 2 --driver-memory 2g --num-executors 2 --queue default --principal hive@MACRO.COM --keytab /spark/conf/hive.keytab --files "/spark/conf/jaas.conf" --driver-java-options "-Djava.security.auth.login.config=/spark/conf/jaas.conf" --conf "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=/spark/conf/jaas.conf" sparkstreaming-1.0-SNAPSHOT.jar

• 注意：如果用yarn运行，/spark/conf下一定要有jaar.conf和hive.keytab等文件配置文件

1. 运行结果如下

1. 运行kafka的producer脚本向kafka2_spark2队列添加数据

1. 数据成功添加

6、Kerberos环境SparkStreaming拉取kafka数据写入

HBase

6.1、测试环境

6.2、环境准备

6.3、程序开发

6.4、测试

6.5、问题

整合运行

• 为了方便运行，所有功能打到一个包下面
• 移动package包到根目录下

1. sparkwordcount运行

   spark2-submit --class Spark_WordCount --master local runPackage/sparkstreaming-1.0-SNAPSHOT.jar file:///package/testText/test.txt

   

2. 非kerberos的spark拉去kafka数据向hive写数据

   spark2-submit --class SparkStreamingFromKafkaToHive_NotKerberos --master local runPackage/sparkstreaming-1.0-SNAPSHOT.jar

3. 非kerberos下kafka模拟生产者发送数据

   sh run_producer.sh MyProducerNotKerberos testText/HiveAndHBaseTestText.txt

   

4. 非kerberos的spark拉去kafka数据向HBase写数据

   spark2-submit --class SparkStreamingFromKafkaToHBase_NotKerberos --master local runPackage/sparkstreaming-1.0-SNAPSHOT.jar

   

5. 非kerberos下kafka模拟生产者发送数据

   sh run_producer.sh MyProducerNotKerberos testText/HiveAndHBaseTestText.txt

   

6. kerberos下spark拉去kafka数据向Hive写数据