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);
        }
    }

}