前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >从 0 到 1 学习Kudu 看这一篇就够了!!

从 0 到 1 学习Kudu 看这一篇就够了!!

作者头像
大数据老哥
发布2021-07-29 16:37:12
2.1K0
发布2021-07-29 16:37:12
举报

前言

最近在招聘要求下突然看到了Apache kudu 于是花了几天时间研究了下,下面简单的给大家介绍下 记得收藏。

一、Kudu 介绍

1.1、背景介绍

在KUDU之前,大数据主要以两种方式存储;

【1】:静态数据

HDFS 引擎作为存储引擎,适用于高吞吐量的离线大数据分析场景。这类存储的局限性是数据无法进行随机的读写。【2】:动态数据HBase、Cassandra 作为存储引擎,适用于大数据随机读写场景。 局限性是批量读取吞吐量远不如 HDFS,不适用于批量数据分析的场景。 从上面分析可知,这两种数据在存储方式上完全不同,进而导致使用场景完全不同,但在真实的场景中,边界可能没有那么清晰,面对既需要随机读写,又需要批量分析的大数据场景,该如何选择呢? 这个场景中,单种存储引擎无法满足业务需求,我们需要通过多种大数据工具组合来满足这一需求,如下图所示:

如上图所示,数据实时写入 HBase,实时的数据更新也在 HBase 完成,为了应对 OLAP 需求,我们定时将 HBase 数据写成静态的文件(如:Parquet)导入到 OLAP 引擎(如:Impala、hive)。这一架构能满足既需要随机读写,又可以支持 OLAP 分析的场景,但他有如下缺点:

  1. 架构复杂。从架构上看,数据在HBase、消息队列、HDFS 间流转,涉及环节太多,运维成本很高。并且每个环节需要保证高可用,都需要维护多个副本,存储空间也有一定的浪费。最后数据在多个系统上,对数据安全策略、监控等都提出了挑战。
  2. 时效性低。数据从HBase导出成静态文件是周期性的,一般这个周期是一天(或一小时),在时效性上不是很高。
  3. 难以应对后续的更新。真实场景中,总会有数据是延迟到达的。如果这些数据之前已经从HBase导出到HDFS,新到的变更数据就难以处理了,一个方案是把原有数据应用上新的变更后重写一遍,但这代价又很高。 为了解决上述架构的这些问题,KUDU应运而生。KUDU的定位是Fast Analytics on Fast Data,是一个既支持随机读写、又支持 OLAP 分析的大数据存储引擎。

从上图可以看出,KUDU 是一个折中的产品,在 HDFS 和 HBase 这两个偏科生中平衡了随机读写和批量分析的性能。从 KUDU 的诞生可以说明一个观点:底层的技术发展很多时候都是上层的业务推动的,脱离业务的技术很可能是空中楼阁。

1.2、Kudu 是什么

官方介绍:Kudu 是为 Apache Hadoop 平台开发的列式存储管理器。Kudu 具有 Hadoop 生态系统应用程序的共同技术属性:它在商品硬件上运行,具有水平可扩展性,并支持高可用操作。简单来说:kudu是一个与Hbase类似的列式存储分布式数据库。

1.3 为什么需要kudu?

HDFS 与HBase的数据存储的缺点目前数据存储有了HDFS与HBase,为什么还要额弄一个kudu呢?HDFS : 使用列式存储格式Apache Parquet , Apache ORC,适合离线分析,不支持单条记录级别的update操作,随机读写能力差HBase :可以进行高效读写,却并不是适合基于SQL的数据分析方向,大批量数据获取的性能差。kudu : 正因为HDFS与HBase有上面这些缺点,kudu较好的解决了HDFS与HBase的这些特点,它不及HDFS批处理快,也不及HBase随机读写能力强,但反过来它比HBase批处理快,而且比HDFS随机读写能力强(适合实时写入或这更新场景频繁的场景).这就是他能解决的问题。

1.4 Kudu 的好处

  • OALP 工作负载的快速处理
  • 与MapReduce 、spark 和其他Hadoop生态系统组件集成
  • 与Apache Impala 紧密集成,使其成为将HDFS于Apache Parquet 结合使用的良好、可变得替代方案。
  • 强大而灵活的一致性模型,允许您根军每个请求选择一致性要求,包括严格序列化一致性的选项。
  • 同时运行顺序和随机工作负载的强大性能
  • 结构化数据模型

二、Apache Kudu架构

  • Table(表) :一张表table是数据存储在kudu的位置。Table具有schema和全局有序的primary key(主键)。Table 被分为很多段,也就是tablets
  • tablet(段) :一个tablet是一张table连续的segment,与其他数据存储引擎或关系型数据库的partition相似。Tablet 存在副本机制,其中一个副本为leader tablet。任何副本都可以对读取进行服务,并且写入是需要在所有副本对应的tablet server之间达成一致性。
  • Tablet server:存储tablet和tablet向client提供服务。对于给定的tablet,一个tablet server 充当leader,其他tablet serverd充当该tablet的follwer副本。只有leader服务写请求,leader与follower为每个服务提供读取请求。
  • Master:主要用来管理元数据(元数据存储只有一个tablet的catalog table中),即tablet 与表的基本信息,监听tserver的状态
  • Catalog Table:元数据表,用来存储table(schema、locations、states)与tablet (现有的tablet列表,每个tablet及其副本所处tserver,tablet当前状态以及开始和结束键)的信息。

2.1 存储结构图

  1. 一个Table包含多个tablet,其中tablet的数量是根据hash或这range进行设置
  2. 一个Table中包含MeteData信息和多个RowSet信息
  3. 一个RowSet中包含一个MemRowSet与0个或多个DiskRowSet,其中MemRowSet存储insert的数据,一个MemRowSet写满会flush到磁盘上 生成一个或多个DiskRowSet,此时MemRowSet清空。MemRowSet默认写满1G或这120sFlush一次(注意:memRowSet是行式存储,DiskRowSet是列式存储,MemRowSet基于primary Key有序)。每隔table中会定期对一些diskrowset做compaction操作,目的是对多个diskRowSet进行重新排序,以此来使其更有序减少diskRowSet的数量,同事在compaction的过程中慧慧resolve掉deltaStores当中的delete记录
  4. 一个DiskRowSet包含baseData与DeltaStores两部分,其中baseData存储的数据看起来不可改变, DeltaStores中存储的是改变得数据
  5. DeltaStores包含一个DeltaMemStores和多个DeltaFile,其中DeltaMemStores放在内存,用来存储update与delete数据,一旦DeltaMemStores写满,会flush成DeltaFile.
  6. 当DeltaFile过多很影响查询性能,所有kudu每个一段时间会执行compaction操作,将其合并到hbaseData中,主要是resolve掉update数据。

2.2 kudu角色分配

  1. kudu主要角色分为master与tserver
  2. master主要负责管理元数据信息,监听server,当server宕机后负责tablet的重新分配
  3. tserver主要负责table的存储和数据的增删改查

2.3 读取流程

客户端将要读取的数据信息发送给master,master对其进行一定的校验,比如表是否存在,字段是否存在,Master返回元数据信息给clinet,然后client与tserver建立连接,通过metaData找到数据所在的rowSet,首先加载内存里面的数据(MemRowSet与DeltMemStore),然后加载磁盘里面的数据,最后返回最终数据给clinet。

  1. client 端master请求查询指定数据
  2. master 对请求进行校验,校验表是否存在,schema中是否存在指定查询的字段,主键是否存在
  3. master通过查询catalog Table返回表,将table对应的tserver信息,tserver状态等元数据信息返回给clinet
  4. clinet与tserver建立连接,通过metaData找到parmary Key 对应的RowSet.
  5. 首先加载RowSet内存中MemRowSet与DeltMemStore中的数据
  6. 然后加载磁盘中的数据,也就是DiskRowSet中BaseDataD与DeltFile中的数据 返回数据给Client
  7. 继续执行4-7的步骤,直到clinet拿所有数据即可

2.4 插入数据

Client首先建立master,获取元数据信息, 然后连接 tserver,查询MemRowSet与DeltMemStroe中是否存在相同primary Key,如果存在,则报错,如果不存在,则将待插入的数据写入WAL日志,然后将数据写入MemRowSet。

  1. client向master请求预跟新表的元数据,首先master会效验表是否存在,字段是否存在,如果效验通过则会返回给client表的分区、tablet tablet所在tserver信息
  2. client 向tserver发起更新请求
  3. 将更新操作预写WAL日志,用来在server宕机后数据恢复
  4. 根据tserver中待更新的数据所处位置不通过,有不同处理方式 数据在内存中,则从MemRowSet中找对数据所处的行,然后在改行的mutaion链表中写入更新消息,在MemRowSet Flush的时候,将数据合并到BaseData中 如果数据在DiskRowSet中,则将更新消息写入DeltMemStore中,DeltMemStore达到一定大小后flush成DeltFile。
  5. 更新完毕后返回消息给client。

三、kudu集群安装

3.1 配置NTP服务

每个节点运行

代码语言:javascript
复制
yum -y install ntp

注释一下四行

代码语言:javascript
复制
#server 0.centos.pool.ntp.org iburst
#server 1.centos.pool.ntp.org iburst
#server 2.centos.pool.ntp.org iburst
#server 3.centos.pool.ntp.org iburst

在第一台服务器node01

代码语言:javascript
复制
vim /etc/ntp.conf

添加如下内容

代码语言:javascript
复制
restrict 192.168.100.0 mask 255.255.255.0 notrap nomodify    
server  node01   prefer                           
server 127.127.1.0                                        
fudge   127.127.1.0 stratum 10

修改node02和node03

代码语言:javascript
复制
vi /etc/ntp.conf

添加如下内容

代码语言:javascript
复制
server  node01 prefer
server  127.127.1.0
fudge   127.127.1.0 stratum 10

启动NTP服务

代码语言:javascript
复制
service ntpd start

chkconfig ntpd on

通过ntpstat 检测服务是否正常

代码语言:javascript
复制
[root@node01 ~]# ntpstat
synchronised to local net (127.127.1.0) at stratum 11
   time correct to within 11 ms
   polling server every 64 s

3.2 kudu安装

每个节点都执行(需要安装包后台回复:kudu 安装包)

代码语言:javascript
复制
 yum install -y  kudu-1.4.0+cdh5.12.2+0-1.cdh5.12.2.p0.8.el7.x86_64.rpm
 yum install -y  kudu-client0-1.4.0+cdh5.12.2+0-1.cdh5.12.2.p0.8.el7.x86_64.rpm
 yum install -y  kudu-client-devel-1.4.0+cdh5.12.2+0-1.cdh5.12.2.p0.8.el7.x86_64.rpm
 yum install -y  kudu-debuginfo-1.4.0+cdh5.12.2+0-1.cdh5.12.2.p0.8.el7.x86_64.rpm
 yum install -y  kudu-master-1.4.0+cdh5.12.2+0-1.cdh5.12.2.p0.8.el7.x86_64.rpm
 yum install -y  kudu-tserver-1.4.0+cdh5.12.2+0-1.cdh5.12.2.p0.8.el7.x86_64.rpm

修改配置 /etc/default/kudu-master (三个节点都需要改下)

代码语言:javascript
复制
export FLAGS_log_dir=/var/log/kudu
export FLAGS_rpc_bind_addresses=node01:7051

修改配置 /etc/default/kudu-tserver

代码语言:javascript
复制
export FLAGS_log_dir=/var/log/kudu
export FLAGS_rpc_bind_addresses=node01:7050

启动服务测试

代码语言:javascript
复制
/etc/init.d/kudu-master start (node01)
Started Kudu Master Server (kudu-master):                  [  确定  ]

/etc/init.d/kudu-tserver start(node01)
Started Kudu Tablet Server (kudu-tserver):                 [  确定  ]

/etc/init.d/kudu-tserver start(node01)
Started Kudu Tablet Server (kudu-tserver):                 [  确定  ]

/etc/init.d/kudu-tserver start(node02)
Started Kudu Tablet Server (kudu-tserver):                 [  确定  ]

通过浏览器查看 http://node01:8051/tablet-servers

四、Java整合Kudu

4.1 添加pom配置

代码语言:javascript
复制
        <dependency>
            <groupId>org.apache.kudu</groupId>
            <artifactId>kudu-client</artifactId>
            <version>1.4.0</version>
        </dependency>

4.2 创建表

代码语言:javascript
复制
private static ColumnSchema newColumn(String name, Type type, Boolean isKey){
        ColumnSchema.ColumnSchemaBuilder column =
                new ColumnSchema.ColumnSchemaBuilder(name, type);
        column.key(isKey);
        return column.build();
    }
    public static void main(String[] args) {
        String masterId = "192.168.100.101";
        KuduClient client = new KuduClient.KuduClientBuilder(masterId).defaultAdminOperationTimeoutMs(6000).build();
        try {
            LinkedList<ColumnSchema> columns  = new LinkedList<>();
            columns.add(newColumn("id", Type.INT32,true));
            columns.add(newColumn("name",Type.STRING,false));
            Schema schema = new Schema(columns);
            CreateTableOptions options = new CreateTableOptions();
            LinkedList<String> parcols  = new LinkedList<>();
            parcols.add("id");
            options.setNumReplicas(1);
            options.addHashPartitions(parcols, 3);
            client.createTable("test", schema, options);
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

4.3 插入数据

代码语言:javascript
复制
 public static void main(String[] args) {
        String masterId = "192.168.100.101";
        KuduClient client = new KuduClient.KuduClientBuilder(masterId).defaultAdminOperationTimeoutMs(6000).build();
        try {
            KuduTable table = client.openTable("test");
            KuduSession kuduSession = client.newSession();
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
            kuduSession.setMutationBufferSpace(3000);
            for (int i = 0; i < 10; i++) {
                Insert insert = table.newInsert();
                insert.getRow().addInt("id", i);
                insert.getRow().addString("name", String.valueOf("name" + i));
                kuduSession.flush();
                kuduSession.apply(insert);
            }
            kuduSession.close();
        }catch (Exception e){
        }
    }

4.4 修改数据

代码语言:javascript
复制
public static void main(String[] args) {
        String masterId = "192.168.100.101";
        KuduClient client = new KuduClient.KuduClientBuilder(masterId).defaultAdminOperationTimeoutMs(6000).build();
        try {
            KuduTable table = client.openTable("test");
            KuduSession kuduSession = client.newSession();
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
            Update update = table.newUpdate();
            PartialRow row = update.getRow();
            row.addInt("id", 1);
            row.addString("name", "大家好");
            kuduSession.apply(update);
            kuduSession.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

4.5 查询数据

代码语言:javascript
复制
    public static void main(String[] args) {
        String masterId = "192.168.100.101";
        KuduClient client = new KuduClient.KuduClientBuilder(masterId).defaultAdminOperationTimeoutMs(6000).build();
        try {
            KuduTable table = client.openTable("test");
            KuduScanner scanner = client.newScannerBuilder(table).build();
            while (scanner.hasMoreRows()) {
                for (RowResult rowResult : scanner.nextRows()) {
                    System.out.println(rowResult.getInt("id") + "\t" + rowResult.getString("name"));
                }
            }
            scanner.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

4.6 删除数据

代码语言:javascript
复制
   public static void main(String[] args) {
        String masterId = "192.168.100.101";
        KuduClient client = new KuduClient.KuduClientBuilder(masterId).defaultAdminOperationTimeoutMs(6000).build();

        try {
            KuduTable table = client.openTable("test");
            KuduSession session = client.newSession();
            Delete delete = table.newDelete();
            delete.getRow().addInt("id", 1);
            session.flush();
            session.apply(delete);
            session.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

五、Spark整合Kudu

5.1 添加pom文件

代码语言:javascript
复制
        <dependency>
            <groupId>org.apache.kudu</groupId>
            <artifactId>kudu-spark2_2.11</artifactId>
            <version>1.4.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.2.0</version>
        </dependency>

5.2 创建表

代码语言:javascript
复制

    val sparkConf = new SparkConf().setAppName("kudu").setMaster("local")

    // 创建spark 环境
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 构建kudu上下文
    val kuduContext = new KuduContext("192.168.100.101", spark.sparkContext)

    val kuduTableName = "spark_kudu_table_student4"
    // 定义schema
    var schema = StructType(
      StructField("id", StringType, false) ::
        StructField("name", StringType, false) ::
        StructField("age", IntegerType, false) :: Nil
    )
    // 定义主键
    val kuduTablePrimaryKey = Seq("id")
    val createTableOptions = new CreateTableOptions()

    createTableOptions.setRangePartitionColumns(Seq("id").asJava)
      .setNumReplicas(3)

    kuduContext.createTable(kuduTableName, schema, kuduTablePrimaryKey, createTableOptions)
     spark.close()
    spark.close()

5.3 删除表

代码语言:javascript
复制
def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("kudu").setMaster("local")
    // 创建spark 环境
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 构建kudu上下文
    val kuduContext = new KuduContext("192.168.100.101", spark.sparkContext)
    val kudu_table_name = "spark_kudu_table_student2"
    val table = kuduContext.tableExists(kudu_table_name)
    if (table) {
      kuduContext.deleteTable(kudu_table_name)
    }
     spark.close()
    spark.close()

  }

5.4 获取数据

代码语言:javascript
复制
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("kudu").setMaster("local")

    // 创建spark 环境
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 构建kudu上下文
    val kuduContext = new KuduContext("192.168.100.101", spark.sparkContext)
    val value: RDD[Row] = kuduContext.kuduRDD(spark.sparkContext, "spark_kudu_table_student4",Seq("id","name","age"))
    value.foreach(println)
     spark.close()
    spark.close()
  }

5.5 插入数据

代码语言:javascript
复制
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("kudu").setMaster("local")

    // 创建spark 环境
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 构建kudu上下文
    val kuduContext = new KuduContext("192.168.100.101", spark.sparkContext)
    import spark.implicits._
    val data = spark.read.text("./data/user.txt").map(row => {
      val arr = row.get(0).toString.split(",")
      (arr(1), arr(0), arr(2).toInt)
    }).toDF("id", "name", "age").dropDuplicates("id")
    kuduContext.insertRows(data, "spark_kudu_table_student4")
     spark.close()
    spark.close()
  }

大数据老哥

希望这篇文章可以帮到你~ 记得点赞收藏哦

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2021-07-05,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 大数据老哥 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 前言
  • 一、Kudu 介绍
    • 1.1、背景介绍
      • 1.2、Kudu 是什么
        • 1.3 为什么需要kudu?
          • 1.4 Kudu 的好处
          • 二、Apache Kudu架构
            • 2.1 存储结构图
              • 2.2 kudu角色分配
                • 2.3 读取流程
                  • 2.4 插入数据
                  • 三、kudu集群安装
                    • 3.1 配置NTP服务
                      • 3.2 kudu安装
                      • 四、Java整合Kudu
                        • 4.1 添加pom配置
                          • 4.2 创建表
                            • 4.3 插入数据
                              • 4.4 修改数据
                                • 4.5 查询数据
                                  • 4.6 删除数据
                                  • 五、Spark整合Kudu
                                    • 5.1 添加pom文件
                                      • 5.2 创建表
                                        • 5.3 删除表
                                          • 5.4 获取数据
                                            • 5.5 插入数据
                                            相关产品与服务
                                            TDSQL MySQL 版
                                            TDSQL MySQL 版(TDSQL for MySQL)是腾讯打造的一款分布式数据库产品,具备强一致高可用、全球部署架构、分布式水平扩展、高性能、企业级安全等特性,同时提供智能 DBA、自动化运营、监控告警等配套设施,为客户提供完整的分布式数据库解决方案。
                                            领券
                                            问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档