目前的项目中需要将kafka队列的数据实时存到hive表中。

1、场景介绍：数据发往kafka，用spark读取kafka的数据，写入到hive表里面（ORC压缩算法，一个分区字段）

2、hive的介绍：hive表是分区表
 

/**
 * SparkStreaming2.3版本 读取kafka 中数据 ：
 *  1.采用了新的消费者api实现，类似于1.6中SparkStreaming 读取 kafka Direct模式。并行度 一样。
 *  2.因为采用了新的消费者api实现，所有相对于1.6的Direct模式【simple api实现】 ，api使用上有很大差别。未来这种api有可能继续变化
 *  3.kafka中有两个参数：
 *      heartbeat.interval.ms：这个值代表 kafka集群与消费者之间的心跳间隔时间，kafka 集群确保消费者保持连接的心跳通信时间间隔。这个时间默认是3s.
 *             这个值必须设置的比session.timeout.ms appropriately 小，一般设置不大于 session.timeout.ms appropriately 的1/3
 *      session.timeout.ms appropriately：
 *             这个值代表消费者与kafka之间的session 会话超时时间，如果在这个时间内，kafka 没有接收到消费者的心跳【heartbeat.interval.ms 控制】，
 *             那么kafka将移除当前的消费者。这个时间默认是10s。
 *             这个时间位于配置 group.min.session.timeout.ms【6s】 和 group.max.session.timeout.ms【300s】之间的一个参数,如果SparkSteaming 批次间隔时间大于5分钟，
 *             也就是大于300s,那么就要相应的调大group.max.session.timeout.ms 这个值。
 *  4.大多数情况下，SparkStreaming读取数据使用 LocationStrategies.PreferConsistent 这种策略，这种策略会将分区均匀的分布在集群的Executor之间。
 *    如果Executor在kafka 集群中的某些节点上，可以使用 LocationStrategies.PreferBrokers 这种策略，那么当前这个Executor 中的数据会来自当前broker节点。
 *    如果节点之间的分区有明显的分布不均，可以使用 LocationStrategies.PreferFixed 这种策略,可以通过一个map 指定将topic分区分布在哪些节点中。
 *
 *  5.新的消费者api 可以将kafka 中的消息预读取到缓存区中，默认大小为64k。默认缓存区在 Executor 中，加快处理数据速度。
 *     可以通过参数 spark.streaming.kafka.consumer.cache.maxCapacity 来增大，也可以通过spark.streaming.kafka.consumer.cache.enabled 设置成false 关闭缓存机制。
 *
 *  6.关于消费者offset
 *    1).如果设置了checkpoint ,那么offset 将会存储在checkpoint中。
 *     这种有缺点: 第一，当从checkpoint中恢复数据时，有可能造成重复的消费，需要我们写代码来保证数据的输出幂等。
 *                第二，当代码逻辑改变时，无法从checkpoint中来恢复offset.
 *    2).依靠kafka 来存储消费者offset,kafka 中有一个特殊的topic 来存储消费者offset。新的消费者api中，会定期自动提交offset。这种情况有可能也不是我们想要的，
 *       因为有可能消费者自动提交了offset,但是后期SparkStreaming 没有将接收来的数据及时处理保存。这里也就是为什么会在配置中将enable.auto.commit 设置成false的原因。
 *       这种消费模式也称最多消费一次，默认sparkStreaming 拉取到数据之后就可以更新offset,无论是否消费成功。自动提交offset的频率由参数auto.commit.interval.ms 决定，默认5s。
 *       如果我们能保证完全处理完业务之后，可以后期异步的手动提交消费者offset.
 *
 *    3).自己存储offset,这样在处理逻辑时，保证数据处理的事务，如果处理数据失败，就不保存offset，处理数据成功则保存offset.这样可以做到精准的处理一次处理数据。
 *注释：这里的解释是北京尚学堂的我敬佩的大佬讲师的
 */
object SparkStreamingOnKafkaDirect {
 def main(args: Array[String]): Unit = {
//    val conf = new SparkConf()
//    conf.setMaster("local")
//    conf.setAppName("SparkStreamingOnKafkaDirect")
   val spark  = SparkSession.builder().appName("test").master("local").enableHiveSupport().getOrCreate()
   val ssc = new StreamingContext(spark.sparkContext,Durations.seconds(3))
   //设置日志级别
   ssc.sparkContext.setLogLevel("Error")

   val kafkaParams = Map[String, Object](
     "bootstrap.servers" -> "node01:9092,node02:9092,node03:9092",
     "key.deserializer" -> classOf[StringDeserializer],
     "value.deserializer" -> classOf[StringDeserializer],
     "group.id" -> "MyGroupId",//

     /**
       * 当没有初始的offset，或者当前的offset不存在，如何处理数据
       *  earliest ：自动重置偏移量为最小偏移量
       *  latest：自动重置偏移量为最大偏移量【默认】
       *  none:没有找到以前的offset,抛出异常
       */
     "auto.offset.reset" -> "earliest",

     /**
       * 当设置 enable.auto.commit为false时，不会自动向kafka中保存消费者offset.需要异步的处理完数据之后手动提交
       */
     "enable.auto.commit" -> (false: java.lang.Boolean)//默认是true
   )

//设置Kafka的topic
   val topics = Array("test")
//创建与Kafka的连接，接收数据
/*这里接收到数据的样子
2019-09-26	1569487411604	1235	497	Kafka	Register
2019-09-26	1569487411604	1235	497	Kafka	Register
2019-09-26	1569487414838	390	   778	Flink	View
*/
   val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
     ssc,
     PreferConsistent,//
     Subscribe[String, String](topics, kafkaParams)
   )
   
//对接收到的数据进行处理，打印出来接收到的key跟value，最后放回的是value
   val transStrem: DStream[String] = stream.map(record => {
     val key_value = (record.key, record.value)
     println("receive message key = "+key_value._1)
     println("receive message value = "+key_value._2)
     key_value._2
   })


//这里用了一下动态创建的Schema
   val structType: StructType = StructType(List[StructField](
     StructField("Date_", StringType, nullable = true),
     StructField("Timestamp_", StringType, nullable = true),
     StructField("UserID", StringType, nullable = true),
     StructField("PageID", StringType, nullable = true),
     StructField("Channel", StringType, nullable = true),
     StructField("Action", StringType, nullable = true)
   ))

//因为foreachRDD可以拿到封装到DStream中的rdd，可以对里面的rdd进行，
/*代码解释：
    先从foreach中拿到一条数据，，在函数map中对接收来的数据用 “\n” 进行切分，放到Row中，用的是动态创建Schema，因为我们需要再将数据存储到hive中，所以需要Schema。
    因为map是transformance算子，所以用rdd.count（）触发一下
     spark.createDataFrame：创建一个DataFrame，因为要注册一个临时表，必须用到DataFrame
     frame.createOrReplaceTempView("t1")：注册临时表
       spark.sql("use spark")：使用 hive 的 spark 库
     result.write.mode(SaveMode.Append).saveAsTable("test_kafka")：将数据放到 test_kafka 中
*/
   transStrem.foreachRDD(one=>{
     val rdd : RDD[Row] = one.map({
       a =>
       val arr = a.toString.split("\t")
 Row(arr(0).toString, arr(1).toString, arr(2).toString, arr(3).toString,arr(4).toString, arr(5).toString)
     })
     rdd.count()
     val frame : DataFrame = spark.createDataFrame(rdd,structType)
//      println(" Scheme: "+frame.printSchema())

     frame.createOrReplaceTempView("t1")
//      spark.sql("select * from t1").show()
     spark.sql("use spark")
     result.write.mode(SaveMode.Append).saveAsTable("test_kafka")
      }
   )

   /**
     * 以上业务处理完成之后，异步的提交消费者offset,这里将 enable.auto.commit 设置成false,就是使用kafka 自己来管理消费者offset
     * 注意这里，获取 offsetRanges: Array[OffsetRange] 每一批次topic 中的offset时，必须从 源头读取过来的 stream中获取，不能从经过stream转换之后的DStream中获取。
     */
   stream.foreachRDD { rdd =>
     val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
     // some time later, after outputs have completed
     stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
   }
   ssc.start()
   ssc.awaitTermination()
   ssc.stop()
 }
}

sparkstreaming 实时读取kafka写入hive优化（高流量）

背景：

kafka流量在800M/s,前任留下的程序大量数据丢失，且逻辑生成复杂，查询hive直接奔溃，优化从两方面，程序优化及小文件合并（生成结果产生大量小文件）

程序直接上代码，
  

def main(args: Array[String]): Unit = {
    val  sdf = new SimpleDateFormat("yyyyMMddHHmm")
    val broker_list = "XXXX";
    val zk = "xxx";
    val confSpark = new SparkConf()
           .setAppName("kafka2hive")
    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .set("spark.rdd.compress", "true")
    .set("spark.sql.shuffle.partitions", "512") //生成的partition根据kafka topic 分区生成，这个配置项貌似没效果
    .set("spark.streaming.stopGracefullyOnShutdown", "true") //能够处理完最后一批数据，再关闭程序，不会发生强制kill导致数据处理中断，没处理完的数据丢失
      .set("spark.streaming.backpressure.enabled","true")//开启后spark自动根据系统负载选择最优消费速率
      .set("spark.shuffle.manager", "sort")
    .set("spark.locality.wait", "5ms")
      //.setMaster("local[*]")

    val kafkaMapParams = Map(
      "auto.offset.reset" -> "largest",
      "group.id" -> "kafka2dhive",
      "zookeeper.session.timeout.ms" -> "40000",
      "metadata.broker.list" -> broker_list,
      "zookeeper.connect" -> zk
    )
    val topicsSet = Set("innerBashData")
    val sc = new SparkContext(confSpark)
    val ssc = new StreamingContext(sc,Seconds(30)) //这个是重点微批处理，根据自己的机器资源，测试调整
    val sqlContext = new HiveContext(sc)
    var daily = sdf.format(new Date()).substring(0,8)
    var dailyTableName = "bashdata"+daily;
    val schema = StructType(
      StructField("ver", StringType, true) ::
        StructField("session_id", StringType, true) ::
        StructField("host_time", StringType, true) ::
        StructField("host_name", StringType, true) ::
        StructField("src_ip", StringType, true) ::
        Nil)

    sqlContext.sql(s"""create table if not exists $dailyTableName(
                  a    string ,
                  b   string ,
                  c   string ,
                  d   string ,
                  e   string 
                 )
                  PARTITIONED BY (hours string,min string)
                  ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
                  STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
                  OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
                     """.stripMargin)

    val lines = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaMapParams, topicsSet)
    lines.foreachRDD( beforerdd => {
      val rdd = beforerdd.map( rdd1 => {
        rdd1._2
      })
      rdd.cache()

     val agentDataFrame = sqlContext.read.schema(schema).json(rdd)
      // .coalesce(10) //控制文件输出个数
      agentDataFrame.registerTempTable("tmp_bashdata")
      sqlContext.sql("set hive.exec.dynamic.partition = true")
      sqlContext.sql("set hive.exec.dynamic.partition.mode = nonstrict")
      sqlContext.sql("set hive.mapred.supports.subdirectories=true")
      sqlContext.sql("set mapreduce.input.fileinputformat.input.dir.recursive=true")
      sqlContext.sql("set mapred.max.split.size=256000000")
      sqlContext.sql("set mapred.min.split.size.per.node=128000000")
      sqlContext.sql("set mapred.min.split.size.per.rack=128000000")
      sqlContext.sql("set hive.merge.mapfiles=true")
      sqlContext.sql("set hive.merge.mapredfiles=true")
      sqlContext.sql("set hive.merge.size.per.task=256000000")
      sqlContext.sql("set hive.merge.smallfiles.avgsize=256000000")
      sqlContext.sql("set hive.groupby.skewindata=true")

      var hours = sdf.format(new Date()).substring(8,10)
      var min = sdf.format(new Date()).substring(10,12)  //每10分钟生成一个文件夹，这tm数据量也够大的
      sqlContext.sql(
        s"""
           |INSERT  OVERWRITE TABLE  $dailyTableName PARTITION(hours='$hours', min='$min')
           |SELECT
           |    a,
           |    b,
           |    c,
           |     d,
           |    e
           |FROM tmp_bashdata
            """.stripMargin)

    });
    ssc.start()
    ssc.awaitTermination()

小文件合并

核心思想是重新生成一张表，指定分区数。脚本如下：

set mapred.reduce.tasks=5;
set mapred.max.split.size=512000000;
insert into table yhtable PARTITION(hours=14,min=1)
select
ver,
session_id,
host_time,
host_name,
src_ip
from 
aa20190624 where hours=14 and min=0;

实现过程如下:以下代码是针对topic的分区只有一个的情况下：

object FamilyBandService {
val logger = LoggerFactory.getLogger(this.getClass)
def main(args: Array[String]): Unit = {
val conf =new SparkConf().setAppName(s"${this.getClass.getSimpleName}").setMaster("local[*]")
conf.set("spark.defalut.parallelism","500")
//每秒钟每个分区kafka拉取消息的速率
      .set("spark.streaming.kafka.maxRatePerPartition","500")
// 序列化
      .set("spark.serilizer","org.apache.spark.serializer.KryoSerializer")
// 建议开启rdd的压缩
      .set("spark.rdd.compress","true")
val sc =new SparkContext(conf)
val ssc =new StreamingContext(sc,Seconds(3))
val brokers = PropertyUtil.getInstance().getProperty("brokerList","")  //配置文件读取工具类需自行编写
val groupId = PropertyUtil.getInstance().getProperty("groupid","")
val topic = PropertyUtil.getInstance().getProperty("topic","")
var topics =Array(topic)
Logger.getLogger("org").setLevel(Level.ERROR) //临时测试的时候只开启error级别，方便排错。
//封装参数
    val kafkaParams =Map[String, Object](
"bootstrap.servers" -> brokers,
"key.deserializer" ->classOf[StringDeserializer],
"value.deserializer" ->classOf[StringDeserializer],
"group.id" -> groupId,
"auto.offset.reset" ->"latest",
"enable.auto.commit" -> (false: java.lang.Boolean))
//单分区情况
//从redis中获取到偏移量
    val offsets: Long = RedisUtil.hashGet("offset","offsets").toLong
val topicPartition: TopicPartition =new TopicPartition(topic,0)
val partitionoffsets:Map[TopicPartition, Long] =Map(topicPartition -> offsets)
//获取到实时流对象
    val kafkaStream =if (offsets ==0) {
KafkaUtils.createDirectStream[String,String](
ssc,
PreferConsistent,  //这里有3种模式,一般情况下，都是使用PreferConsistent
//LocationStrategies.PreferConsistent：将在可用的执行器之间均匀分配分区。
//PreferBrokers  执行程序与Kafka代理所在的主机相同，将更喜欢在该分区的Kafka leader上安排分区
//PreferFixed 如果您在分区之间的负载有显着偏差，这允许您指定分区到主机的显式映射（任何未指定的分区将使用一致的位置）。
Subscribe[String,String](topics, kafkaParams) //消息订阅
)
}else {
KafkaUtils.createDirectStream[String,String](
ssc,
PreferConsistent,
Subscribe[String,String](topics, kafkaParams, partitionoffsets)  //此种方式是针对具体某个分区或者topic只有一个分区的情况
)
}
//业务处理
    kafkaStream.foreachRDD(rdd => {
val ranges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  //获取到分区和偏移量信息
val events: RDD[Some[String]] = rdd.map(x => {
val data = x.value()
Some(data)
})
val session = SQLContextSingleton.getSparkSession(events.sparkContext)  //构建一个Sparksession的单例
session.sql("set hive.exec.dynamic.partition=true")      //配置hive支持动态分区
session.sql("set hive.exec.dynamic.partition.mode=nonstrict")   //配置hive动态分区为非严格模式
//如果将数据转换为Seq(xxxx),然后倒入隐式转换import session.implicalit._  是否能实现呢，答案是否定的。
val dataRow = events.map(line => {                                           //构建row
val temp = line.get.split("###")                                                   
Row(temp(0), temp(1), temp(2), temp(3), temp(4), temp(5))
})
//"deviceid","code","time","info","sdkversion","appversion"
 val structType =StructType(Array(                          //确定字段的类别
StructField("deviceid", StringType,true),
StructField("code", StringType,true),
StructField("time", StringType,true),
StructField("info", StringType,true),
StructField("sdkversion", StringType,true),
StructField("appversion", StringType,true)
))
val df = session.createDataFrame(dataRow, structType)   //构建df
df.createOrReplaceTempView("jk_device_info")
session.sql("insert into test.jk_device_info select * from jk_device_info")
for (rs <- ranges) {
//实时保存偏移量到redis
        val value = rs.untilOffset.toString
RedisUtil.hashSet("offset","offsets", value)   //偏移量保存
println(s"the offset:${value}")
}
})
println("xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")
ssc.start()
ssc.awaitTermination()
}
}
 

//多分区情况
val partitions = 3
 var fromdbOffset =Map[TopicPartition, Long]()
for (partition <-0 until partitions) {
val topicPartition =new TopicPartition(topic, partition)
val offsets = RedisUtil.hashGet("offset",s"${topic}_${partition}").toLong
fromdbOffset += (topicPartition -> offsets)
}
//获取到实时流对象
    val kafkaStream =if (fromdbOffset.size==0) {
KafkaUtils.createDirectStream[String,String](
ssc,
PreferConsistent,
Subscribe[String,String](topics, kafkaParams)
)
}else {
KafkaUtils.createDirectStream[String,String](
ssc,
PreferConsistent,
//        Subscribe[String, String](topics, kafkaParams, partitionoffsets) 订阅具体某个分区
        ConsumerStrategies.Assign[String,String](fromdbOffset.keys, kafkaParams, fromdbOffset)
)
}
 

SQLContextSingleton单例

def getSparkSession(sc: SparkContext): SparkSession = {
if (sparkSession ==null) {
sparkSession = SparkSession
.builder()
.enableHiveSupport()
.master("local[*]")
.config(sc.getConf)
.config("spark.files.openCostInBytes", PropertyUtil.getInstance().getProperty("spark.files.openCostInBytes"))
//連接到hive元數據庫
      .config("hive.metastore.uris","thrift://192.168.1.61:9083")
//--files hdfs:///user/processuser/hive-site.xml 集群上運行需要指定hive-site.xml的位置
      .config("spark.sql.warehouse.dir","hdfs://192.168.1.61:8020/user/hive/warehouse")
.getOrCreate()
}
sparkSession
}

如果需要连接到hive必须要注意的几个事项：

1，指定hive的元数据地址

2，指定spark.sql.warehouse.dir的数据存储位置

3，enableHiveSupport()

4，resource下要放hive-site.xml文件

xml文件需要配置的信息，以下信息均可从集群的配置中得到：

hive.exec.scratchdir

hive.metastore.warehouse.dir

hive.querylog.location

hive.metastore.uris    

javax.jdo.option.ConnectionURL

javax.jdo.option.ConnectionDriverName

javax.jdo.option.ConnectionUserName

javax.jdo.option.ConnectionPassword

5，本地执行要指定hadoop的目录

System.setProperty("hadoop.home.dir", PropertyUtil.getInstance().getProperty("localMode"))

#hadoop info

localMode=D://hadoop-2.6.5//hadoop-2.6.5

clusterMode=file://usr//hdp//2.6.2.0-205//hadoop

 

读取Hive

    //如果想让hive运行在spark上，一定要开启spark对hive的支持
    val spark = SparkSession.builder()
      .appName("HiveOnSpark")
      .master("local[*]")
      .enableHiveSupport() //启用spark对hive的支持(可以兼容hive的语法了)
      .getOrCreate()

    //想要使用hive的元数据库，必须指定hive元数据的位置，添加一个hive-site.xml到当前程序的classpath下即可
    val result: DataFrame = spark.sql("SELECT * FROM users ORDER BY id DESC")
//    val sql: DataFrame = spark.sql("CREATE TABLE niu (id bigint, name string)")
//    sql.show()