1.将hdfs-site,core-site.hive-site文件拷贝到resources目录下

image.png

2.添加maven依赖

org.apache.spark

spark-streaming_2.11

2.1.1

org.apache.spark

spark-streaming-kafka-0-10_2.11

2.1.1

org.apache.hive

hive-jdbc

1.2.1

org.apache.hive

hive-service

1.2.1

org.apache.spark

spark-hive_2.11

2.1.1

mysql

mysql-connector-java

5.1.27

3.编写代码

object KafkaDemo {

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

//1.创建 SparkConf 并初始化 SSC

val sparkConf: SparkConf = new SparkConf()

.setMaster("local[*]")

.setAppName("KafkaSparkStreaming")

val ssc = new StreamingContext(sparkConf, Seconds(20))

//2.定义 kafka 参数

val brokers = "s201:9092"

val consumerGroup = "spark"

//3.将 kafka 参数映射为 map

val kafkaParams = Map[String, String](

"bootstrap.servers" -> brokers,

"group.id" -> consumerGroup,

"key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",

"value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"

)

//要监听的Topic，可以同时监听多个

val topics = Array("student")

//4.通过 KafkaUtil 创建 kafkaDSteam

val dstream = KafkaUtils.createDirectStream(ssc, LocationStrategies.PreferConsistent,

ConsumerStrategies.Subscribe[String, String](topics, kafkaParams))

dstream.foreachRDD(rdd => {

//获取到分区和偏移量信息

val ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

val events: RDD[Some[String]] = rdd.map(x => {

val data = x.value()

Some(data)

})

val warehouseLocation = "spark-warehouse"

val spark = SparkSession

.builder()

.appName("Spark Hive Example")

.enableHiveSupport()

.config("spark.sql.warehouse.dir", warehouseLocation)

.config("user.name", "hadoop")

.config("HADOOP_USER_NAME", "hive")

.getOrCreate()

import spark.sql

//配置hive支持动态分区

sql("set hive.exec.dynamic.partition=true")

//配置hive动态分区为非严格模式

sql("set hive.exec.dynamic.partition.mode=nonstrict")

//如果将数据转换为Seq(xxxx),然后倒入隐式转换import session.implicalit._ 是否能实现呢，答案是否定的。

//构建row

val dataRow = events.map(line => {

val temp = line.get.split("###")

Row(temp(0), temp(1), temp(2))

})

//"deviceid","code","time","info","sdkversion","appversion"

//确定字段的类别

val structType = StructType(Array(

StructField("name", StringType, true),

StructField("age", StringType, true),

StructField("major", StringType, true)

))

//构建df

val df = spark.createDataFrame(dataRow, structType)

val unit = df.createOrReplaceTempView("jk_device_info")

val frame = sql("insert into myhive.student select * from jk_device_info")

})

//6.启动 SparkStreaming

ssc.start()

ssc.awaitTermination()

}

}

启动hadoop,zookeeper,kafka

/opt/module/hadoop-2.7.2/sbin/start-dfs.sh

/opt/module/hadoop-2.7.2/sbin/start-yarn.sh

zk.sh start

#! /bin/bash

case $1 in

"start"){

for i in s201 s202 s203

do

ssh $i "/opt/module/zookeeper-3.4.10/bin/zkServer.sh start"

done

};;

"stop"){

for i in s201 s202 s203

do

ssh $i "/opt/module/zookeeper-3.4.10/bin/zkServer.sh stop"

done

};;

"status"){

for i in s201 s202 s203

do

ssh $i "/opt/module/zookeeper-3.4.10/bin/zkServer.sh status"

done

};;

esac

kf.sh start

#! /bin/bash

case $1 in

"start"){

for i in s201 s202 s203

do

echo " --------启动 $i Kafka-------"

# 用于KafkaManager监控

ssh $i "export JMX_PORT=9988 && /opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/modu

le/kafka/config/server.properties " done

};;

"stop"){

for i in s201 s202 s203

do

echo " --------停止 $i Kafka-------"

ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh stop"

done

};;

esac

kafka发送消息

bin/kafka-console-producer.sh --broker-list s201:9092 --topic student

xiekai###24###ningdu

hive中查看是否是否插入

xiekain 24 ningdu

image.png

插入成功

注意，运行是可能会报HDFS的权限问题，所以需要加入运行时参数

image.png