Kudu的介绍及使用

心已赠人 2023-05-28 05:19 47阅读 0赞

前文：

过往采用Hive的离线处理时效性低，计算任务过于集中，查询效率低。SparkStreaming+Hive的数据清洗线使得多套数据流过于复杂。未来的数据仓库场景越来越趋向于实时数仓。

一、引入

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70][]

二、架构图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 1][]

2、架构及数据量

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 2][]

3、文件结构

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 3][]

4、目录结构

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 4][]

5、读写流程及分区策略

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 5][]

三、分区及策略

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 6][]

四、代码

1、通过SparkSql操作数据

object kudu_02_Main {
      def main(args: Array[String]): Unit = {
        val spark = SparkSession.builder()
          .master("local[*]")
          .appName("kudu_02_Main")
          .getOrCreate()
    
        val df = spark.read.options(Map("kudu.master" -> "kudu-01:7051",
          "kudu.table" -> "impala::test.my_table_kudu")).format("org.apache.kudu.spark.kudu").load()//.kudu
    
        df.show();
    
        /**
         * +------+---------------------+---------+
         * | name | time                | message |
         * +------+---------------------+---------+
         * | joey | 2018-01-01 00:00:00 | hello   |
         * | ross | 2018-02-01 00:00:00 | goodbye |
         * +------+---------------------+---------+
         */
    
        df.select("name").where("name = 'joey'").show()
    
        println("######################################################################")
    
    //    //注册全局视窗
    //    df.createGlobalTempView("my_table_kudu1")
    //    spark.sql("select * from global_temp.my_table_kudu1").show()
    
        //注册临时视窗
        df.createOrReplaceTempView("my_table_kudu2")
        spark.sql("select * from my_table_kudu2 where name = 'joey'").show()
    
        /**
         * create table test.my_tb2_kudu(
         * name string,
         * age int,
         * city string,
         * primary key(name)
         * )
         * partition by hash(name) partitions 2
         * stored as kudu;
         */
    
        val students: Array[Student] = Array(
          Student("jack", 30, "new york"),
          Student("jason", 20, "beijing"),
          Student("cesar", 18, "guangzhou"),
          Student("anderson", 25, "dongguang")
        )
    
        import spark.implicits._
        val stuRDD: RDD[Student] = spark.sparkContext.parallelize(students)
        val stuDF = stuRDD.toDF()
    
        stuDF.write
          .options(Map("kudu.master"-> "bt-03:7051", "kudu.table"-> "impala::test.my_tb2_kudu"))
          .mode("append")
          .format("org.apache.kudu.spark.kudu").save
    
      }
    }

2、通过kuduContext操作数据

object kudu_03_Main {
    
      case class Student(name: String, age: Int, city: String)
    
      case class delStu(name:String)
    
      def main(args: Array[String]): Unit = {
    
        val spark = SparkSession.builder()
          .master("local[*]")
          .appName("kudu_03_Main")
          .getOrCreate()
        val kuduContext = new KuduContext("kudu-01:7051", spark.sparkContext)
        println(kuduContext.tableExists("impala::test.my_table_kudu"))
    
        /**
         * 建表语句
         * create table test.my_tb3_kudu(
         * name string,
         * age int,
         * city string,
         * primary key(name)
         * )
         * partition by hash(name) partitions 2
         * stored as kudu;
         */
    
        //增加
        val students: Array[Student] = Array(
          Student("jack", 30, "new york"),
          Student("jason", 20, "beijing"),
          Student("cesar", 18, "guangzhou"),
          Student("anderson", 25, "dongguang")
        )
        //RDD与DF互相转换
        import spark.implicits._
        val studentsRDD: RDD[Student] = spark.sparkContext.parallelize(students)
        val df: DataFrame = studentsRDD.toDF()
        df.show()
        //kuduContext.insertRows(df,"impala::test.my_tb3_kudu")
    
        val new_stu: Array[Student] = Array(
          Student("jack", 31, "dongbei"),
          Student("jason", 21, "shanghai")
        )
        val new_setRDD: RDD[Student] = spark.sparkContext.parallelize(new_stu)
        val df4 = new_setRDD.toDF()
        kuduContext.upsertRows(df4,"impala::test.my_tb3_kudu")
    
    
        //删除
        val delStus: Seq[delStu] = Seq(
          delStu("jason"),
          delStu("cesar")
        )
        //RDD与DF互相转换
        import spark.implicits._
        val delStusRDD: RDD[delStu] = spark.sparkContext.parallelize(delStus)
        val df1: DataFrame = delStusRDD.toDF()
        df1.show()
        //kuduContext.deleteRows(df1,"impala::test.my_tb3_kudu")
    
    
        val df2 = spark.read.options(Map("kudu.master" -> "bt-03:7051",
          "kudu.table" -> "impala::test.my_tb3_kudu")).format("org.apache.kudu.spark.kudu").load()
        df2.show()
    
        df2.createOrReplaceTempView("my_tb3_kudu")
        val df3: DataFrame = spark.sql("select name from my_tb3_kudu where name = 'anderson'")
        //kuduContext.deleteRows(df3,"impala::test.my_tb3_kudu")
    
      }
    
    }

五、Kudu和HBase的对比

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 7][]

六、配置

master.gflagfile

# Do not modify these two lines. If you wish to change these variables,
    # modify them in /etc/default/kudu-master.
    --fromenv=rpc_bind_addresses
    --fromenv=log_dir
    
    --fs_wal_dir=/data/kudu/wal/master
    --fs_data_dirs=/data/kudu/data/master
    --master_addresses=kudu-01:7051,kudu-02:7051,kudu-03:7051
    -rpc_encryption=disabled
    -rpc_authentication=disabled
    -trusted_subnets=0.0.0.0/0
    
    
    --rpc_num_acceptors_per_address=5
    --master_ts_rpc_timeout_ms=60000
    --rpc_service_queue_length=1000

tserver.gflagfile

# Do not modify these two lines. If you wish to change these variables,
    # modify them in /etc/default/kudu-tserver.
    --fromenv=rpc_bind_addresses
    --fromenv=log_dir
    
    --fs_wal_dir=/data/kudu/wal/server
    --fs_data_dirs=/data/kudu/data/server
    --tserver_master_addrs=kudu-01:7051,kudu-02:7051,kudu-03:7051
    -rpc_encryption=disabled
    -rpc_authentication=disabled
    -trusted_subnets=0.0.0.0/0
    
    #性能调优
    --memory_limit_hard_bytes=27917287424
    --block_cache_capacity_mb=7680
    --maintenance_manager_num_threads=3
    
    --rpc_num_acceptors_per_address=5
    --master_ts_rpc_timeout_ms=60000
    --rpc_service_queue_length=1000

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70]: /images/20230528/8e095f55b58a4d2498f2f382801b314c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 1]: /images/20230528/e29e628f4f944b91bcec68f49b15b118.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 2]: /images/20230528/1ca255371700481da6c2e354fd6f22b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 3]: /images/20230528/984b3de99d104e8e873d024067a74b4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 4]: /images/20230528/6050a2800de14c85b7dc51d1651fc004.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 5]: /images/20230528/5fde33b0c8be4485b33b914fa6a72d31.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 6]: /images/20230528/cdc7cc0ef9f34334adfca162659f7066.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY4NzA3NA_size_16_color_FFFFFF_t_70 7]: /images/20230528/de0c6a89d93a4bbdb0a17572f0c13507.png