SparkStreaming读Kafka数据写Kudu

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。

Fayson的github: https://github.com/fayson/cdhproject

提示:代码块部分可以左右滑动查看噢

1.文档编写目的


在前面的文章Fayson介绍过《SparkStreaming读HBase写HDFS》及《SparkingStreaming读Kafka写Kudu》,本篇文章Fayson主要介绍使用Scala语言开发一个SparkStreaming应用读取Kafka数据并写入Kudu。本文的数据流图如下:

  • 内容概述

1.环境准备

2.编写SparkSteaming代码读取Kafka数据并写入Kudu

3.流程测试

4.总结

  • 测试环境

1.CM和CDH版本为5.12.1

2.采用root用户操作

  • 前置条件

1.集群已安装Kafka

2.集群已安装Kudu且正常运行

3.集群未启用Kerberos

2.环境准备


1.编写向Kafka生成数据的ReadUserInfoFIleToKafka.java代码,具体内容可以在Fayson的GitHub上查看

https://github.com/fayson/cdhproject/blob/master/kafkademo/src/main/java/com/cloudera/nokerberos/ReadUserInfoFIleToKafka.java
https://github.com/fayson/cdhproject/tree/master/kafkademo/0283-kafka-shell

(可左右滑动)

2.使用mvn命令将编写好的代码编译打包封装成脚本

mvn clean package

使用mvn命令将工程依赖包导出到lib目录

mvn dependency:copy-dependencies -DoutputDirectory=/Users/fayson/Desktop/lib

(可左右滑动)

编写run.sh脚本

#!/bin/bash

#########################################
# 创建Topic
# kafka-topics --create --zookeeper cdh01.fayson.com:2181,cdh02.fayson.com:2181,cdh03.fayson.com:2181 --replication-factor 3 --partitions 3 --topic kafka_sparkstreaming_kudu_topic
#
########################################
JAVA_HOME=/usr/java/jdk1.8.0_131
#要读取的文件
read_file=$1
for file in `ls lib/*jar`
do
    CLASSPATH=$CLASSPATH:$file
done
export CLASSPATH
${JAVA_HOME}/bin/java -Xms1024m -Xmx2048m com.cloudera.nokerberos.ReadUserInfoFIleToKafka $read_file

(可左右滑动)

准备测试数据ods_user_600.txt

封装好的脚本目录结构如下:

将打包好的jar包拷贝至lib目录下。

3. 创建用于测试的Kafka Topic

kafka-topics --create --zookeeper cdh01.fayson.com:2181,cdh02.fayson.com:2181,cdh03.fayson.com:2181 --replication-factor 3 --partitions 3 --topic kafka_sparkstreaming_kudu_topic

(可左右滑动)

4. 通过CM配置SparkKudu依赖包kudu-spark_2.10-1.4.0-cdh5.12.1.jar,将依赖包部署至CDH集群所有节点的/opt/cloudera/parcels/CDH/jars目录,然后通过CM配置Spark GateWay的spark-env.sh配置

export SPARK_DIST_CLASSPATH=$SPARK_DIST_CLASSPATH:/opt/cloudera/parcels/CDH/spark-streaming-kafka_2.10-1.6.0-cdh5.12.1.jar
export SPARK_DIST_CLASSPATH=$SPARK_DIST_CLASSPATH:/opt/cloudera/parcels/CDH/jars/kudu-spark_2.10-1.4.0-cdh5.12.1.jar

(可左右滑动)

还需要依赖spark-streaming-kafka_2.10-1.6.0-cdh5.12.1.jar包,保存并重新部署客户端配置。

3.编写SparkStreaming写Kudu示例


1.使用Maven创建Scala工程,工程依赖pom文件

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka_2.10</artifactId>
    <version>1.6.0-cdh5.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.10</artifactId>
    <version>1.6.0-cdh5.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-assembly_2.10</artifactId>
    <version>1.6.0-cdh5.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.kudu</groupId>
    <artifactId>kudu-spark_2.10</artifactId>
    <version>1.4.0-cdh5.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.kudu</groupId>
    <artifactId>kudu-client</artifactId>
    <version>1.4.0-cdh5.12.1</version>
</dependency>

(可左右滑动)

2.在resource目录下创建0285.properties文件,内容如下

kafka.brokers=cdh01.fayson.com:9092,cdh02.fayson.com:9092,cdh03.fayson.com:9092
kafka.topics=kafka_sparkstreaming_kudu_topic
kudumaster.list=cdh01.fayson.com:7051,cdh02.fayson.com:7051,cdh03.fayson.com:7051

(可左右滑动)

3. 新建Kafka2Spark2Kudu.scala文件,示例代码如下:

package com.cloudera.streaming
import java.io.{File, FileInputStream}
import scala.collection.JavaConversions._
import java.util.Properties
import kafka.serializer.StringDecoder
import org.apache.commons.lang3.StringUtils
import org.apache.kudu.client.CreateTableOptions
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.sql.types._
import org.apache.log4j.{Level, Logger}
import scala.util.parsing.json.JSON
/**
  * package: com.cloudera.streaming
  * describe: SparkStreaming 应用实时读取Kafka数据,解析后存入Kudu
  * 使用spark-submit的方式提交作业
    spark-submit --class com.cloudera.streaming.Kafka2Spark2Kudu \
    --master yarn-client --num-executors 3 --driver-memory 1g \
    --driver-cores 1 --executor-memory 1g --executor-cores 1 \
    spark-demo-1.0-SNAPSHOT.jar
  * creat_user: Fayson 
  * email: htechinfo@163.com
  * creat_date: 2018/5/28
  * creat_time: 上午10:09
  * 公众号:Hadoop实操
  */
object Kafka2Spark2Kudu {
  Logger.getLogger("org").setLevel(Level.ERROR) //设置日志级别
  var confPath: String = System.getProperty("user.dir") + File.separator + "conf/0285.properties"
  /**
    * 建表Schema定义
    */
  val userInfoSchema = StructType(
      //         col name   type     nullable?
      StructField("id", StringType , false) ::
      StructField("name" , StringType, true ) ::
      StructField("sex" , StringType, true ) ::
      StructField("city" , StringType, true ) ::
      StructField("occupation" , StringType, true ) ::
      StructField("tel" , StringType, true ) ::
      StructField("fixPhoneNum" , StringType, true ) ::
      StructField("bankName" , StringType, true ) ::
      StructField("address" , StringType, true ) ::
      StructField("marriage" , StringType, true ) ::
      StructField("childNum", StringType , true ) :: Nil
  )
  /**
    * 定义一个UserInfo对象
    */
  case class UserInfo (
    id: String,
    name: String,
    sex: String,
    city: String,
    occupation: String,
    tel: String,
    fixPhoneNum: String,
    bankName: String,
    address: String,
    marriage: String,
    childNum: String
  )
  def main(args: Array[String]): Unit = {
    //加载配置文件
    val properties = new Properties()
    val file = new File(confPath)
    if(!file.exists()) {
      System.out.println(Kafka2Spark2Kudu.getClass.getClassLoader.getResource("0285.properties"))
      val in = Kafka2Spark2Kudu.getClass.getClassLoader.getResourceAsStream("0285.properties")
      properties.load(in);
    } else {
      properties.load(new FileInputStream(confPath))
    }
    val brokers = properties.getProperty("kafka.brokers")
    val topics = properties.getProperty("kafka.topics")
    val kuduMaster = properties.getProperty("kudumaster.list")
    println("kafka.brokers:" + brokers)
    println("kafka.topics:" + topics)
    println("kudu.master:" + kuduMaster)
    if(StringUtils.isEmpty(brokers)|| StringUtils.isEmpty(topics) || StringUtils.isEmpty(kuduMaster)) {
      println("未配置Kafka和KuduMaster信息")
      System.exit(0)
    }
    val topicsSet = topics.split(",").toSet
    val sparkConf = new SparkConf().setAppName("Kafka2Spark2Kudu")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(5)) //设置Spark时间窗口,每5s处理一次
    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
    val dStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)
    //引入隐式
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._
    val kuduContext = new KuduContext(kuduMaster, sc)
    //判断表是否存在
    if(!kuduContext.tableExists("user_info")) {
      println("create Kudu Table :{user_info}")
      val createTableOptions = new CreateTableOptions()
      createTableOptions.addHashPartitions(List("id"), 8).setNumReplicas(3)
      kuduContext.createTable("user_info", userInfoSchema, Seq("id"), createTableOptions)
    }
    dStream.foreachRDD(rdd => {
      //将rdd数据重新封装为Rdd[UserInfo]
      val newrdd = rdd.map(line => {
        val jsonObj =  JSON.parseFull(line._2)
        val map:Map[String,Any] = jsonObj.get.asInstanceOf[Map[String, Any]]
        new UserInfo(
          map.get("id").get.asInstanceOf[String],
          map.get("name").get.asInstanceOf[String],
          map.get("sex").get.asInstanceOf[String],
          map.get("city").get.asInstanceOf[String],
          map.get("occupation").get.asInstanceOf[String],
          map.get("mobile_phone_num").get.asInstanceOf[String],
          map.get("fix_phone_num").get.asInstanceOf[String],
          map.get("bank_name").get.asInstanceOf[String],
          map.get("address").get.asInstanceOf[String],
          map.get("marriage").get.asInstanceOf[String],
          map.get("child_num").get.asInstanceOf[String]
        )
      })
      //将RDD转换为DataFrame
      val userinfoDF = sqlContext.createDataFrame(newrdd)
      kuduContext.upsertRows(userinfoDF, "user_info")
    })
    ssc.start()
    ssc.awaitTermination()
  }
}

(可左右滑动)

3. 使用mvn命令将编写好的SparkStreaming代码打包,注意由于工程中有scala代码在编译是命令中需要加scala:compile

mvn clean scala:compile package

(可左右滑动)

4.流程测试


1. 将编译好的SparkStreaming应用Jar包上传至有Spark Gateway节点的服务器上

conf/0285.properties内容如下:

2. 使用spark-submit命令提交SparkStreaming作业

spark-submit --class com.cloudera.streaming.Kafka2Spark2Kudu \
  --master yarn-client --num-executors 3 --driver-memory 1g \
  --driver-cores 1 --executor-memory 1g --executor-cores 1 \
  spark-demo-1.0-SNAPSHOT.jar

(可左右滑动)

通过CM查看SparkStreaming作业是否正常运行

Yarn的8088界面查看

3. Spark作业启动成功后自动创建Kudu的user_info表

进入表明显页面,找到Impala建表语句

CREATE EXTERNAL TABLE `user_info` STORED AS KUDU
TBLPROPERTIES(
  'kudu.table_name' = 'user_info',
  'kudu.master_addresses' = 'cdh01.fayson.com:7051,cdh02.fayson.com:7051,cdh03.fayson.com:7051');

(可左右滑动)

在Hue中执行建表

表中无数据

4. 运行脚本向Kafka生产数据

[root@cdh01 0283-kafka-shell]# cd /root/0283-kafka-shell
[root@cdh01 0283-kafka-shell]# sh run.sh ods_user_600.txt

(可左右滑动)

5. 通过Hue查看Kudu的user_info表数据

Kafka的数据已成功的录入到Kudu的user_info表中

统计写入的数据量为600,与写入Kafka的数据一致

5.总结


1.由于Spark中默认没有Spark-Streaming-Kafka的依赖包,需要将相应的依赖包添加到/opt/cloudera/parcels/CDH/jars目录下,然后在spark-env.sh中配置相应的依赖包路径,否则会报类找不到的异常。

2.访问Kudu使用的是kudu-spark_2.10-1.4.0-cdh5.12.1.jar,需要将该依赖包添加到/opt/cloudera/parcels/CDH/jars目录下,并配置spark-env.sh的环境变量

GitHub地址:

https://github.com/fayson/cdhproject/blob/master/sparkdemo/src/main/scala/com/cloudera/streaming/Kafka2Spark2Kudu.scala

提示:代码块部分可以左右滑动查看噢

为天地立心,为生民立命,为往圣继绝学,为万世开太平。 温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。

推荐关注Hadoop实操,第一时间,分享更多Hadoop干货,欢迎转发和分享。

原创文章,欢迎转载,转载请注明:转载自微信公众号Hadoop实操

原文发布于微信公众号 - Hadoop实操(gh_c4c535955d0f)

原文发表时间:2018-06-02

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏芋道源码1024

分布式作业系统 Elastic-Job-Lite 源码分析 —— 作业执行

Lite调度作业( LiteJob ),作业被调度后,调用 #execute() 执行作业。

6532
来自专栏个人分享

Shuffle相关分析

 Shuffle描述是一个过程,表现出的是多对多的依赖关系。Shuffle是连接map阶段和Reduce阶段的纽带,每个Reduce Task都会从Map Ta...

1084
来自专栏Hadoop实操

Spark2Streaming读Kerberos环境的Kafka并写数据到Hive

在前面的文章Fayson介绍了一些关于Spark2Streaming的示例如《Spark2Streaming读Kerberos环境的Kafka并写数据到HBas...

1.1K3
来自专栏祝威廉

Spark 2.0 Structured Streaming 分析

Spark 2.0 将流式计算也统一到DataFrame里去了,提出了Structured Streaming的概念,将数据源映射为一张无线长度的表,同时将流式...

1503
来自专栏Jed的技术阶梯

Kafka 0.8.2.2 Producer报错:java.net.ConnectException: Connection timed out: no further information

1.1K2
来自专栏岑玉海

Spark Streaming编程指南

Overview Spark Streaming属于Spark的核心api,它支持高吞吐量、支持容错的实时流数据处理。 它可以接受来自Kafka, Flume,...

7255
来自专栏Hadoop实操

Spark2Streaming读Kerberos环境的Kafka并写数据到HDFS

在前面的文章Fayson介绍了一些关于Spark2Streaming的示例如《Spark2Streaming读Kerberos环境的Kafka并写数据到HBas...

3081
来自专栏Albert陈凯

Spark系列课程-00xxSpark RDD持久化

我们这节课讲一下RDD的持久化 ? RDD的持久化 这段代码我们上午已经看过了,有瑕疵大家看出来了吗? 有什么瑕疵啊? 大家是否还记得我在第二节课的时候跟大...

4128
来自专栏个人分享

Hive架构及Hive On Spark

(1)Table:每个表都对应在HDFS中的目录下,数据是经过序列化后存储在该目录中。同时Hive也支持表中的数据存储在其他类型的文件系统中,如NFS或本地文件...

4462
来自专栏行者悟空

Spark RDD的Action

1556

扫码关注云+社区

领取腾讯云代金券