kafka 的pom文件_基于SparkStreaming+Kafka+HBase实时点击流案例

大数据技术与架构点击右侧关注，大数据开发领域最强公众号！

暴走大数据点击右侧关注，暴走大数据！

背景

Kafka实时记录从数据采集工具Flume或业务系统实时接口收集数据，并作为消息缓冲组件为上游实时计算框架提供可靠数据支撑，Spark 1.3版本后支持两种整合Kafka机制(Receiver-based Approach 和 Direct Approach)，具体细节请参考文章最后官方文档链接，数据存储使用HBase

实现思路

实现Kafka消息生产者模拟器
Spark-Streaming采用Direct Approach方式实时获取Kafka中数据
Spark-Streaming对数据进行业务计算后数据存储到HBase

本地虚拟机集群环境配置

由于笔者机器性能有限，hadoop/zookeeper/kafka集群都搭建在一起主机名分别为hadoop1,hadoop2,hadoop3; hbase为单节点在hadoop1

缺点及不足

代码设计上有些许缺陷，比如spark-streaming计算后数据保存hbase逻辑性能待优化。代码实现Kafka消息模拟器

package clickstreamimport java.util.{Properties, Random, UUID}import kafka.producer.{KeyedMessage, Producer, ProducerConfig}import org.codehaus.jettison.json.JSONObject

object KafkaMessageGenerator {private val random = new Random()private var pointer = -1private val os_type = Array("Android", "IPhone OS","None", "Windows Phone")

  def click() : Double = {    random.nextInt(10)  }

  def getOsType() : String = {    pointer = pointer + 1if(pointer >= os_type.length) {      pointer = 0os_type(pointer)    } else {os_type(pointer)    }  }

  def main(args: Array[String]): Unit = {    val topic = "user_events"//本地虚拟机ZK地址    val brokers = "hadoop1:9092,hadoop2:9092,hadoop3:9092"    val props = new Properties()    props.put("metadata.broker.list", brokers)    props.put("serializer.class", "kafka.serializer.StringEncoder")

    val kafkaConfig = new ProducerConfig(props)    val producer = new Producer[String, String](kafkaConfig)

while(true) {// prepare event data      val event = new JSONObject()      event        .put("uid", UUID.randomUUID())//随机生成用户id        .put("event_time", System.currentTimeMillis.toString) //记录时间发生时间        .put("os_type", getOsType) //设备类型        .put("click_count", click) //点击次数

// produce event message      producer.send(new KeyedMessage[String, String](topic, event.toString))println("Message sent: " + event)

      Thread.sleep(200)    }  }}

Spark-Streaming主类

package clickstreamimport kafka.serializer.StringDecoderimport net.sf.json.JSONObjectimport org.apache.hadoop.hbase.client.{HTable, Put}import org.apache.hadoop.hbase.util.Bytesimport org.apache.hadoop.hbase.{HBaseConfiguration, TableName}import org.apache.spark.SparkConfimport org.apache.spark.streaming.kafka.KafkaUtilsimport org.apache.spark.streaming.{Seconds, StreamingContext}

object PageViewStream {  def main(args: Array[String]): Unit = {var masterUrl = "local[2]"if (args.length > 0) {      masterUrl = args(0)    }

// Create a StreamingContext with the given master URL    val conf = new SparkConf().setMaster(masterUrl).setAppName("PageViewStream")    val ssc = new StreamingContext(conf, Seconds(5))

// Kafka configurations    val topics = Set("PageViewStream")//本地虚拟机ZK地址    val brokers = "hadoop1:9092,hadoop2:9092,hadoop3:9092"    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,"serializer.class" -> "kafka.serializer.StringEncoder")

// Create a direct stream    val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

    val events = kafkaStream.flatMap(line => {      val data = JSONObject.fromObject(line._2)Some(data)    })// Compute user click times    val userClicks = events.map(x => (x.getString("uid"), x.getInt("click_count"))).reduceByKey(_ + _)    userClicks.foreachRDD(rdd => {      rdd.foreachPartition(partitionOfRecords => {        partitionOfRecords.foreach(pair => {//Hbase配置          val tableName = "PageViewStream"          val hbaseConf = HBaseConfiguration.create()          hbaseConf.set("hbase.zookeeper.quorum", "hadoop1:9092")          hbaseConf.set("hbase.zookeeper.property.clientPort", "2181")          hbaseConf.set("hbase.defaults.for.version.skip", "true")//用户ID          val uid = pair._1//点击次数          val click = pair._2//组装数据          val put = new Put(Bytes.toBytes(uid))          put.add("Stat".getBytes, "ClickStat".getBytes, Bytes.toBytes(click))          val StatTable = new HTable(hbaseConf, TableName.valueOf(tableName))          StatTable.setAutoFlush(false, false)//写入数据缓存          StatTable.setWriteBufferSize(3*1024*1024)          StatTable.put(put)//提交          StatTable.flushCommits()        })      })    })    ssc.start()    ssc.awaitTermination()

  }

}

Maven POM文件

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0modelVersion>

<groupId>com.guofei.sparkgroupId><artifactId>RiskControlartifactId><version>1.0-SNAPSHOTversion><packaging>jarpackaging>

<name>RiskControlname><url>http://maven.apache.orgurl>

<properties><project.build.sourceEncoding>UTF-8project.build.sourceEncoding>properties>

<dependencies>

<dependency><groupId>org.apache.sparkgroupId><artifactId>spark-core_2.10artifactId><version>1.3.0version>dependency><dependency><groupId>org.apache.sparkgroupId><artifactId>spark-streaming_2.10artifactId><version>1.3.0version>dependency>

<dependency><groupId>org.apache.sparkgroupId><artifactId>spark-streaming-kafka_2.10artifactId><version>1.3.0version>dependency>

<dependency><groupId>org.apache.hbasegroupId><artifactId>hbaseartifactId><version>0.96.2-hadoop2version><type>pomtype>dependency>

<dependency><groupId>org.apache.hbasegroupId><artifactId>hbase-serverartifactId><version>0.96.2-hadoop2version>dependency><dependency><groupId>org.apache.hbasegroupId><artifactId>hbase-clientartifactId><version>0.96.2-hadoop2version>dependency><dependency><groupId>org.apache.hbasegroupId><artifactId>hbase-commonartifactId><version>0.96.2-hadoop2version>dependency><dependency><groupId>commons-iogroupId><artifactId>commons-ioartifactId><version>1.3.2version>dependency><dependency><groupId>commons-logginggroupId><artifactId>commons-loggingartifactId><version>1.1.3version>dependency><dependency><groupId>log4jgroupId><artifactId>log4jartifactId><version>1.2.17version>dependency>

<dependency><groupId>com.google.protobufgroupId><artifactId>protobuf-javaartifactId><version>2.5.0version>dependency><dependency><groupId>io.nettygroupId><artifactId>nettyartifactId><version>3.6.6.Finalversion>dependency><dependency><groupId>org.apache.hbasegroupId><artifactId>hbase-protocolartifactId><version>0.96.2-hadoop2version>dependency><dependency><groupId>org.apache.zookeepergroupId><artifactId>zookeeperartifactId><version>3.4.5version>dependency><dependency><groupId>org.cloudera.htracegroupId><artifactId>htrace-coreartifactId><version>2.01version>dependency><dependency><groupId>org.codehaus.jacksongroupId><artifactId>jackson-mapper-aslartifactId><version>1.9.13version>dependency><dependency><groupId>org.codehaus.jacksongroupId><artifactId>jackson-core-aslartifactId><version>1.9.13version>dependency><dependency><groupId>org.codehaus.jacksongroupId><artifactId>jackson-jaxrsartifactId><version>1.9.13version>dependency><dependency><groupId>org.codehaus.jacksongroupId><artifactId>jackson-xcartifactId><version>1.9.13version>dependency><dependency><groupId>org.slf4jgroupId><artifactId>slf4j-apiartifactId><version>1.6.4version>dependency><dependency><groupId>org.slf4jgroupId><artifactId>slf4j-log4j12artifactId><version>1.6.4version>dependency>

<dependency><groupId>org.apache.hadoopgroupId><artifactId>hadoop-clientartifactId><version>2.6.4version>dependency><dependency><groupId>commons-configurationgroupId><artifactId>commons-configurationartifactId><version>1.6version>dependency><dependency><groupId>org.apache.hadoopgroupId><artifactId>hadoop-authartifactId><version>2.6.4version>dependency><dependency><groupId>org.apache.hadoopgroupId><artifactId>hadoop-commonartifactId><version>2.6.4version>dependency>

<dependency><groupId>net.sf.json-libgroupId><artifactId>json-libartifactId><version>2.4version><classifier>jdk15classifier>dependency>

<dependency><groupId>org.codehaus.jettisongroupId><artifactId>jettisonartifactId><version>1.1version>dependency>

<dependency><groupId>redis.clientsgroupId><artifactId>jedisartifactId><version>2.5.2version>dependency><dependency><groupId>org.apache.commonsgroupId><artifactId>commons-pool2artifactId><version>2.2version>dependency>dependencies>

<build><sourceDirectory>src/main/scalasourceDirectory><testSourceDirectory>src/test/scalatestSourceDirectory><plugins><plugin><groupId>net.alchim31.mavengroupId><artifactId>scala-maven-pluginartifactId><version>3.2.2version><executions><execution><goals><goal>compilegoal><goal>testCompilegoal>goals><configuration><args><arg>-make:transitivearg><arg>-dependencyfilearg><arg>${project.build.directory}/.scala_dependenciesarg>args>configuration>execution>executions>plugin>

<plugin><groupId>org.apache.maven.pluginsgroupId><artifactId>maven-shade-pluginartifactId><version>2.4.3version><executions><execution><phase>packagephase><goals><goal>shadegoal>goals><configuration><filters><filter><artifact>*:*artifact><excludes><exclude>META-INF/*.SFexclude><exclude>META-INF/*.DSAexclude><exclude>META-INF/*.RSAexclude>excludes>filter>filters>configuration>execution>executions>plugin>plugins>build>project>

FAQ

Maven导入json-lib报错Failure to find net.sf.json-lib:json-lib:jar:2.3 inhttp://repo.maven.apache.org/maven2 was cached in the localrepository解决：http://stackoverflow.com/questions/4173214/maven-missing-net-sf-json-libnet.sf.json-libjson-lib2.4jdk15
执行Spark-Streaming程序报错org.apache.spark.SparkException: Task not serializable

userClicks.foreachRDD(rdd => {rdd.foreachPartition(partitionOfRecords => {partitionOfRecords.foreach(这里面的代码中所包含的对象必须是序列化的这里面的代码中所包含的对象必须是序列化的这里面的代码中所包含的对象必须是序列化的})})})

执行Maven打包报错，找不到依赖的jar包error：not found: object kafkaERROR import kafka.javaapi.producer.Producer解决：win10本地系统用户/xxx/.m2/ 目录含有中文

参考文档

spark-streaming官方文档http://spark.apache.org/docs/latest/streaming-programming-guide.html
spark-streaming整合kafka官方文档http://spark.apache.org/docs/latest/streaming-kafka-integration.html
spark-streaming整合flume官方文档http://spark.apache.org/docs/latest/streaming-flume-integration.html
spark-streaming整合自定义数据源官方文档http://spark.apache.org/docs/latest/streaming-custom-receivers.html
spark-streaming官方scala案例https://github.com/apache/spark/tree/master/examples/src/main/scala/org/apache/spark/examples/streaming
简单之美博客http://shiyanjun.cn/archives/1097.html

作者：MichaelFly链接：https://www.jianshu.com/p/ccba410462ba欢迎点赞+收藏+转发朋友圈素质三连

文章不错？点个【在看】吧！ ?