Kudu的安装和使用

1、下载rpm

　　地址：https://archive.cloudera.com/cdh5/redhat/7/x86_64/cdh/5/RPMS/x86_64/

　　下载所有kudu开头的rpm包

2、安装：
　　rpm -ivh * --nodeps

3、启动：
　　service kudu-master start
　　service kudu-tserver start

4、指令：

     查看集群
　　kudu cluster ksck localhost

     查看master状态或flag
　　kudu master status localhost
　　kudu master get_flags localhost
　　可以查看到bound_http_addresses的port，则访问地址：localhost:port 则是web管理页面。

     查看tserver状态或flag
　　kudu tserver status localhost
　　kudu tserver get_flags localhost

     查看表
　　kudu table list 172.22.19.143

5、DDL、DML
　　• 若使用impala，可以直接执行create sql。
　　• 若使用Java API，可以直接连接kudu并创建表、增删改查操作。

6、用户
　　kudu没有用户的概念，需要依赖 kerberos进行实现。

7、表权限设置
kudu没有用户、角色、权限等概念，需要用 CDH的sentry+impala实现用户权限的分配。

[cdh01-dev:21000]>grant select on table a_kudu_database.a_kudu_table to role read;
[cdh01-dev:21000]>grant all on table b_kudu_database.b_kudu_table to role ddl;
[cdh01-dev:21000]>grant role read to group a_user;
[cdh01-dev:21000]>grant role ddl to group b_user;
[cdh01-dev:21000]>

8、Java API 连接Kudu

由于kerberos和impala都比较重，暂时不进行安装，直接使用Java api的方式直接访问kudu，执行DDL、DML等操作。

注意：修改客户端的hosts文件（Kudu服务器的hostname为Icloud_143，ip为172.22.19.143），则需要在客户端hosts添加：
172.22.19.143   Icloud_143

否则会报connect fail等问题。

示例代码

package com.example.kudu;

import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import org.junit.Before;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

@SpringBootTest
class KuduApplicationTests {

    private KuduClient kuduClient;

    private String kuduMaster;

    private String tableName;

    @Before
    public void init() {
        try{
            System.out.println("initing........................");
            //初始化操作
            kuduMaster = "172.22.19.143:7051";
            //指定表名
            tableName = "student";
            KuduClient.KuduClientBuilder kuduClientBuilder = new KuduClient.KuduClientBuilder(kuduMaster);
            kuduClientBuilder.defaultOperationTimeoutMs(1800000);
            kuduClient = kuduClientBuilder.build();
            System.out.println("服务器地址" + kuduMaster + ":客户端"+ kuduClient +"初始化成功...");
        }catch (Exception e){
            System.out.println(e);
        }
    }

    private static ColumnSchema newColumn(String name, Type type, boolean iskey) {
        ColumnSchema.ColumnSchemaBuilder column = new ColumnSchema.ColumnSchemaBuilder(name, type);
        column.key(iskey);
        return column.build();
    }
    /**
     * 创建无分区表
     */
    @Test
    public void createTable() throws KuduException {
        init();

        // 设置表的schema
        List<ColumnSchema> columns = new LinkedList<ColumnSchema>();
        columns.add(newColumn("id", Type.STRING, true));
        columns.add(newColumn("name", Type.STRING, false));
        columns.add(newColumn("age", Type.INT32, false));
        columns.add(newColumn("sex", Type.INT32, false));
        Schema schema = new Schema(columns);

        // 设置表的replica备份和分区规则
        List<String> parcols = new LinkedList<String>();
        parcols.add("id");

        //创建表时提供的所有选项
        CreateTableOptions options = new CreateTableOptions();
        options.setNumReplicas(1);  //设置表的备份数
        options.setRangePartitionColumns(parcols);  //设置range分区
        options.addHashPartitions(parcols, 3);  //设置hash分区和数量
        try {
            kuduClient.createTable("student",schema, options);
        } catch (KuduException e) {
            e.printStackTrace();
        }finally {
            if (kuduClient != null){
                kuduClient.close();
            }
        }
    }



    /**
     * 向表加载数据
     */
    @Test
    public void insertTable() throws KuduException {
        init();
        //向表加载数据需要一个 kuduSession 对象
        KuduSession kuduSession = kuduClient.newSession();
//        kuduSession.set
        kuduSession.setTimeoutMillis(100000);
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
        //需要使用 kuduTable 来构建 Operation 的子类实例对象
        KuduTable kuduTable = kuduClient.openTable(tableName);
        for (int i = 1; i <= 10; i++) {
            Insert insert = kuduTable.newInsert();
            PartialRow row = insert.getRow();
            row.addString("id", i+"");
            row.addString("name", "zhangsan-" + i);
            row.addInt("age", 20 + i);
            row.addInt("sex", i % 2);
            //最后实现执行数据的加载操作
            kuduSession.apply(insert);
        }
    }




    /**
     * 查询表的数据结果
     */
    @Test
    public void queryData() throws KuduException {
        init();
        //构建一个查询的扫描器
        KuduScanner.KuduScannerBuilder kuduScannerBuilder =
                kuduClient.newScannerBuilder(kuduClient.openTable(tableName));
        ArrayList<String> columnsList = new ArrayList<String>();
        columnsList.add("id");
        columnsList.add("name");
        columnsList.add("age");
        columnsList.add("sex");
        kuduScannerBuilder.setProjectedColumnNames(columnsList);
        //返回结果集
        KuduScanner kuduScanner = kuduScannerBuilder.build();
        //遍历
        while (kuduScanner.hasMoreRows()) {
            RowResultIterator rowResults = kuduScanner.nextRows();
            while (rowResults.hasNext()) {
                RowResult row = rowResults.next();
                String id = row.getString("id");
                String name = row.getString("name");
                int age = row.getInt("age");
                int sex = row.getInt("sex");
                System.out.println(">>>>>>>>>>  id=" + id + " name=" + name + " age=" + age + "sex = " + sex);
            }
        }
    }




    /**
     * 修改表的数据
     */
    @Test
    public void updateData() throws KuduException {
        init();
        //修改表的数据需要一个 kuduSession 对象
        KuduSession kuduSession = kuduClient.newSession();
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
        //需要使用 kuduTable 来构建 Operation 的子类实例对象
        KuduTable kuduTable = kuduClient.openTable(tableName);
        //Update update = kuduTable.newUpdate();
        //如果 id 存在就表示修改，不存在就新增
        Upsert upsert = kuduTable.newUpsert();
        PartialRow row = upsert.getRow();
        row.addInt("id", 100);
        row.addString("name", "zhangsan-100");
        row.addInt("age", 100);
        row.addInt("sex", 0);
        //最后实现执行数据的修改操作
        kuduSession.apply(upsert);
    }




    /**
     * 删除数据
     */
    @Test
    public void deleteData() throws KuduException {
        init();
        //删除表的数据需要一个 kuduSession 对象
        KuduSession kuduSession = kuduClient.newSession();
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
        //需要使用 kuduTable 来构建 Operation 的子类实例对象
        KuduTable kuduTable = kuduClient.openTable(tableName);
        Delete delete = kuduTable.newDelete();
        PartialRow row = delete.getRow();
        row.addInt("id", 100);
        kuduSession.apply(delete);//最后实现执行数据的删除操作
    }




    @Test
    public void dropTable() throws KuduException {
        init();
        if (kuduClient.tableExists(tableName)) {
            kuduClient.deleteTable(tableName);
        }
    }

}