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

温馨提示:如果使用电脑查看图片不清晰,可以使用手机打开文章单击文中的图片放大查看高清原图。

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

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

1.文档编写目的


在前面的文章Fayson介绍了在Kerberos环境下《Spark2Streaming读Kerberos环境的Kafka并写数据到Kudu》,本篇文章Fayson主要介绍如何使用Spark2 Streaming访问非Kerberos环境的Kafka并将接收到的数据写入Kudu。

  • 文章概述

1.环境准备

2.Spark2Streaming示例开发

3.示例运行

4.总结

  • 测试环境

1.CM和CDH版本为5.15

2.CDK2.2.0(Apache Kafka0.10.2)

3.Spark2.2.0

4.操作系统版本为RedHat7.4

2.环境准备


1.准备向Kakfa发送数据的脚本,关于脚本这里就不在过多的介绍前面很多文章都有介绍,具体可以参考Fayson的GitHub:

https://github.com/fayson/cdhproject/tree/master/kafkademo/0283-kafka-shell

根据需要将conf下面的配置文件修改为自己集群的环境即可,发送至Kafka的JSON数据示例如下:

{
   "occupation": "生产工作、运输工作和部分体力劳动者",
   "address": "台东东二路16号-8-8",
   "city": "长治",
   "marriage": "1",
   "sex": "1",
   "name": "仲淑兰",
   "mobile_phone_num": "13607268580",
   "bank_name": "广州银行31",
   "id": "510105197906185179",
   "child_num": "1",
   "fix_phone_num": "15004170180"
}

(可左右滑动)

2.登录CM进入SPARK2服务的配置项将spark_kafka_version的kafka版本修改为0.10

修改完成后并部署客户端配置

3.Spark2Streaming示例代码


1.使用maven创建scala语言的spark2demo工程,pom.xml依赖如下

<dependency>
    <groupId>org.apache.kudu</groupId>
    <artifactId>kudu-spark2_2.11</artifactId>
    <version>1.6.0-cdh5.14.2</version>
</dependency>
<dependency>
    <groupId>org.apache.kudu</groupId>
    <artifactId>kudu-client</artifactId>
    <version>1.6.0-cdh5.14.2</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.2.0.cloudera2</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId>
    <version>2.2.0.cloudera2</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.11</artifactId>
    <version>2.2.0.cloudera2</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
    <version>2.2.0.cloudera2</version>
</dependency>
<dependency>
    <groupId>org.scala-lang</groupId>
    <artifactId>scala-library</artifactId>
    <version>2.11.8</version>
</dependency>

(可左右滑动)

具体需要的依赖包,可以参考Fayson前面的文章《Spark2Streaming读Kerberos环境的Kafka并写数据到Kudu

2.在resources下创建0294.properties配置文件,内容如下:

kafka.brokers=cdh02.fayson.com:9092,cdh03.fayson.com:9092,cdh04.fayson.com:9092
kafka.topics=kafka_kudu_topic
kudumaster.list=cdh01.fayson.com,cdh02.fayson.com,cdh03.fayson.com

(可左右滑动)

3.创建Kafka2Spark2Kudu.scala类

package com.cloudera.streaming.nokerberos

import java.io.{File, FileInputStream}
import java.util.Properties

import org.apache.commons.lang.StringUtils
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.kudu.client.CreateTableOptions
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies}

import scala.util.parsing.json.JSON
import scala.collection.JavaConverters._

/**
  * package: com.cloudera.streaming.nokerberos
  * 使用spark2-submit的方式提交作业
    spark2-submit --class com.fayson.streaming.nokerberos.Kafka2Spark2Kudu \
    --master yarn \
    --deploy-mode client \
    --executor-memory 2g \
    --executor-cores 2 \
    --driver-memory 2g \
    --num-executors 2 \
    spark2-demo-1.0-SNAPSHOT.jar
  * creat_user: Fayson
  * email: htechinfo@163.com
  * creat_date: 2018/8/6
  * creat_time: 下午5:05
  * 公众号:Hadoop实操
  */
object Kafka2Spark2Kudu {

  Logger.getLogger("com").setLevel(Level.ERROR) //设置日志级别

  var confPath: String = System.getProperty("user.dir") + File.separator + "conf/0294.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("0294.properties"))
      val in = Kafka2Spark2Kudu.getClass.getClassLoader.getResourceAsStream("0294.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 spark = SparkSession.builder().appName("Kafka2Spark2Kudu-nokerberos").config(new SparkConf()).getOrCreate()
    val ssc = new StreamingContext(spark.sparkContext, Seconds(5)) //设置Spark时间窗口,每5s处理一次
    val kafkaParams = Map[String, Object]("bootstrap.servers" -> brokers
      , "auto.offset.reset" -> "latest"
      , "key.deserializer" -> classOf[StringDeserializer]
      , "value.deserializer" -> classOf[StringDeserializer]
      , "group.id" -> properties.getProperty("group.id")
    )

    val dStream = KafkaUtils.createDirectStream[String, String](ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topicsSet, kafkaParams))

    //引入隐式
    import spark.implicits._
    val kuduContext = new KuduContext(kuduMaster, spark.sparkContext)

    //判断表是否存在
    if(!kuduContext.tableExists("user_info")) {
      println("create Kudu Table :{user_info}")
      val createTableOptions = new CreateTableOptions()
      createTableOptions.addHashPartitions(List("id").asJava, 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.value())
        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 = spark.sqlContext.createDataFrame(newrdd)
      kuduContext.upsertRows(userinfoDF, "user_info")
    })
    ssc.start()
    ssc.awaitTermination()
  }
}

(可左右滑动)

4.使用mvn命令编译工程,注意由于是scala工程编译时mvn命令要加scala:compile

mvn clean scala:compile package

5.将编译好的spark2-demo-1.0-SNAPSHOT.jar包及配置文件上传至服务器

0294.properties配置文件内容如下:

4.示例运行


1.使用spark2-submit命令向集群提交Spark2Streaming作业

spark2-submit --class com.cloudera.streaming.nokerberos.Kafka2Spark2Kudu \
    --master yarn \
    --deploy-mode client \
    --executor-memory 2g \
    --executor-cores 2 \
    --driver-memory 2g \
    --num-executors 2 \
    spark2-demo-1.0-SNAPSHOT.jar

(可左右滑动)

通过CM查看作业是否提交成功

Spark2的UI界面

2.查看Kudu Master的UI界面,Tables列表可以看到user_info表已被创建

找到Kudu向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')

(可左右滑动)

3.运行脚本向Kafka的kafka_kudu_topic生产消息

4.通过Hue查看数据是否已插入Kudu表

5.总结


1.本示例中Spark2Streaming读取非Kerberos环境的Kafka集群,使用的是spark-streaming-kafka0.10.0版本的依赖包,在Spark中提供两个的另外一个版本的为0.8.0版本,在选择依赖包时需要注意与Spark版本的兼容性问题,具体可以参考官网地址:

http://spark.apache.org/docs/2.2.0/streaming-kafka-integration.html

2.检查/opt/cloudera/parcels/SPARK2/lib/spark2/jars目录下是否有其它版本的spark-streaming-kafka的依赖包,如果存在需要删除,否则会出现版本冲突问题。

3.Spark2默认的kafka版本为0.9需要通过CM将默认的Kafka版本修改为0.10

GitHub地址如下:

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

https://github.com/fayson/cdhproject/blob/master/spark2demo/src/main/resources/0294.properties

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

为天地立心,为生民立命,为往圣继绝学,为万世开太平。 温馨提示:如果使用电脑查看图片不清晰,可以使用手机打开文章单击文中的图片放大查看高清原图。

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

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

原文发表时间:2018-08-07

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏绿巨人专栏

Spark集群 + Akka + Kafka + Scala 开发(4) : 开发一个Kafka + Spark的应用

3517
来自专栏Hadoop实操

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

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

1.1K3
来自专栏木东居士的专栏

用MPI进行分布式内存编程(入门篇)

4543
来自专栏JavaEdge

史上最快! 10小时大数据入门实战(五)-分布式计算框架MapReduce1 MapReduce概述2 MapReduce编程模型之通过wordcount词频统计分析案例入门MapReduce执行流程

1933
来自专栏祝威廉

Spark 2.0 Structured Streaming 分析

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

1513
来自专栏行者悟空

Spark RDD的Action

1566
来自专栏Jed的技术阶梯

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

1.1K2
来自专栏xingoo, 一个梦想做发明家的程序员

Structured Streaming教程(1) —— 基本概念与使用

在有过1.6的streaming和2.x的streaming开发体验之后,再来使用Structured Streaming会有一种完全不同的体验,尤其是在代码设...

1511
来自专栏Albert陈凯

3.4 RDD的计算

3.4 RDD的计算 3.4.1 Ta s k简介 原始的RDD经过一系列转换后,会在最后一个RDD上触发一个动作,这个动作会生成一个Job。在Job被划分为...

37010
来自专栏Albert陈凯

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

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

4128

扫码关注云+社区

领取腾讯云代金券