sparkStreaming连接kafka整合hbase和redis

sparkStreaming消费kafka数据，并将数据保存到redis和hbase当中去，实现实时

import org.apache.hadoop.hbase.client.{Admin, Connection}
import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedisobject StreamingKafka  extends Logging{def main(args: Array[String]): Unit = {val brokers = ConfigUtil.getConfig(Constants.KAFKA_BOOTSTRAP_SERVERS)val topics = Array(ConfigUtil.getConfig(Constants.CHENG_DU_GPS_TOPIC),ConfigUtil.getConfig(Constants.HAI_KOU_GPS_TOPIC))val conf = new SparkConf().setMaster("local[1]").setAppName("sparkKafka")val group:String = "gps_consum_group"val kafkaParams = Map[String, Object]("bootstrap.servers" -> brokers,"key.deserializer" -> classOf[StringDeserializer],"value.deserializer" -> classOf[StringDeserializer],"group.id" -> group,"auto.offset.reset" -> "latest",// earliest,latest,和none"enable.auto.commit" -> (false: java.lang.Boolean))val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()val context: SparkContext = sparkSession.sparkContextcontext.setLogLevel("WARN")// val streamingContext = new StreamingContext(conf,Seconds(5))//获取streamingContextval streamingContext: StreamingContext =  new StreamingContext(context,Seconds(1))val result: InputDStream[ConsumerRecord[String, String]] = HbaseTools.getStreamingContextFromHBase(streamingContext,kafkaParams,topics,group,"(.*)gps_topic")result.foreachRDD(eachRdd =>{if(!eachRdd.isEmpty()){eachRdd.foreachPartition(eachPartition =>{val connection: Connection = HBaseUtil.getConnectionval jedis: Jedis = JedisUtil.getJedis//判断表是否存在，如果不存在就进行创建val admin: Admin = connection.getAdminif(!admin.tableExists(TableName.valueOf(Constants.HTAB_GPS))){val htabgps = new HTableDescriptor(TableName.valueOf(Constants.HTAB_GPS))htabgps.addFamily(new HColumnDescriptor(Constants.DEFAULT_FAMILY))admin.createTable(htabgps)}if(!admin.tableExists(TableName.valueOf(Constants.HTAB_HAIKOU_ORDER))){val htabgps = new HTableDescriptor(TableName.valueOf(Constants.HTAB_HAIKOU_ORDER))htabgps.addFamily(new HColumnDescriptor(Constants.DEFAULT_FAMILY))admin.createTable(htabgps)}eachPartition.foreach(record =>{//保存到HBase和redisval consumerRecords: ConsumerRecord[String, String] = HbaseTools.saveToHBaseAndRedis(connection,jedis, record)})JedisUtil.returnJedis(jedis)connection.close()})//更新offsetval offsetRanges: Array[OffsetRange] = eachRdd.asInstanceOf[HasOffsetRanges].offsetRanges//result.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)  //将offset提交到默认的kafka的topic里面去保存for(eachrange <-  offsetRanges){val startOffset: Long = eachrange.fromOffset  //起始offsetval endOffset: Long = eachrange.untilOffset  //结束offsetval topic: String = eachrange.topicval partition: Int = eachrange.partitionHbaseTools.saveBatchOffset(group,topic,partition+"",endOffset)}}})streamingContext.start()streamingContext.awaitTermination()}
}

sparkStreaming连接kafka整合hbase和redis相关推荐

flink连接kafka整合hbase，scala
解析kafka当中的json格式的数据,入hbase import java.util.Propertiesimport com.alibaba.fastjson.{JSON, JSONObject} ...
2021年大数据Spark（四十二）：SparkStreaming的Kafka快速回顾与整合说明
目录 Kafka快速回顾消息队列: 发布/订阅模式: Kafka 重要概念: 常用命令整合说明两种方式两个版本API 在实际项目中,无论使用Storm还是SparkStreaming与Flin ...
2.4-2.5、Hive整合（整合Spark、整合Hbase）、连接方式Cli、HiveServer和hivemetastore、Squirrel SQL Client等
2.4其它整合 2.4.1Hive整合Spark Spark整合hive,需要将hive_home下的conf下的hive_site.xml放到spark_home下的conf目录下.(3台服务器都做 ...
flink入门3-Flink连接Kafka、Redis，实现Kafka Source/Redis Sink
本篇文章将会一步步实现如何使用Flink对接Kafka和Redis,并将Kafka中的数据存储到Redis中,这种场景也是真实项目会遇到的. 1.单机部署Kafka 1.1 下载Kafka压缩包,解压 ...
Maven+Eclipse+SparkStreaming+Kafka整合
2019独角兽企业重金招聘Python工程师标准>>> 版本号: maven3.5.0 scala IDE for Eclipse:版本(4.6.1) spark-2. ...
大数据Hadoop、Hive、Kafka、Hbase、Spark等框架面经
大数据组件学习路线: 阶段1:学习绿色箭头的知识点: 阶段2:学习红色箭头的知识点: 阶段3:学习蓝色箭头的知识点: 1 Hadoop 1.1 Hadoop1.x与Hadoop2.x的区别 1.2 ...
SpringBoot整合HBase将数据写入Docker中的HBase
在之前的项目里,docker容器中已经运行了HBase,现将API操作HBase实现数据的增删改查通过SpringBoot整合Hbase是一个很好的选择首先打开IDEA,创建项目(project) ...
阿里P8架构师谈：MongoDB、Hbase、Redis等NoSQL优劣势、应用场景
NoSQL的四大种类 NoSQL数据库在整个数据库领域的江湖地位已经不言而喻.在大数据时代,虽然RDBMS很优秀,但是面对快速增长的数据规模和日渐复杂的数据模型,RDBMS渐渐力不从心,无法应对很多数 ...
apache druid 与kafka整合使用
前言在上一篇,我们了解了apache druid的搭建,以及如何快速导入外部数据源到apache druid中进行数据分析和使用本篇,我们结合一个实际的简单的应用场景,来说说apache drui ...

sparkStreaming连接kafka整合hbase和redis

sparkStreaming连接kafka整合hbase和redis相关推荐

最新文章

热门文章