Java代碼操作Kudu

一、構建maven工程

二、匯入依賴

三、???????創建包結構

四、???????初始化方法

五、???????創建表

六、???????插入資料

七、???????查詢資料

八、修改資料

九、???????洗掉資料

十、???????修改表

十一、???????洗掉表

Java代碼操作Kudu

一、???????構建maven工程

二、匯入依賴

<repositories>
    <repository>
        <id>cloudera</id>
        <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
    </repository>
</repositories>

<dependencies>
    <dependency>
        <groupId>org.apache.kudu</groupId>
        <artifactId>kudu-client</artifactId>
        <version>1.9.0-cdh6.2.1</version>
    </dependency>

    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
    </dependency>

    <dependency>
        <groupId>org.apache.kudu</groupId>
        <artifactId>kudu-client-tools</artifactId>
        <version>1.9.0-cdh6.2.1</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.kudu/kudu-spark2 -->
    <dependency>
        <groupId>org.apache.kudu</groupId>
        <artifactId>kudu-spark2_2.11</artifactId>
        <version>1.9.0-cdh6.2.1</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>2.1.0</version>
    </dependency>
</dependencies>

三、???????創建包結構

包名	說明
cn.it	代碼所在的包目錄

四、???????初始化方法

package cn.it;

import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Type;
import org.apache.kudu.client.KuduClient;
import org.junit.Before;

public class TestKudu {
    //定義KuduClient客戶端物件
    private static KuduClient kuduClient;
    //定義表名
    private static String tableName = "person";

    /**
     * 初始化方法
     */
    @Before
    public void init() {
        //指定master地址
        String masterAddress = "node2.cn";
        //創建kudu的資料庫連接
        kuduClient = new KuduClient.KuduClientBuilder(masterAddress).defaultSocketReadTimeoutMs(6000).build();
    }

    //構建表schema的欄位資訊
    //欄位名稱   資料型別     是否為主鍵
    public ColumnSchema newColumn(String name, Type type, boolean isKey) {
        ColumnSchema.ColumnSchemaBuilder column = new ColumnSchema.ColumnSchemaBuilder(name, type);
        column.key(isKey);
        return column.build();
    }
}

五、???????創建表

/**  使用junit進行測驗
 *
 * 創建表
 * @throws KuduException
 */
@Test
public void createTable() throws KuduException {
    //設定表的schema
    List<ColumnSchema> columns = new LinkedList<ColumnSchema>();
    columns.add(newColumn("CompanyId", Type.INT32, true));
    columns.add(newColumn("WorkId", Type.INT32, false));
    columns.add(newColumn("Name", Type.STRING, false));
    columns.add(newColumn("Gender", Type.STRING, false));
    columns.add(newColumn("Photo", Type.STRING, false));
    Schema schema = new Schema(columns);
    //創建表時提供的所有選項
    CreateTableOptions tableOptions = new CreateTableOptions();
    //設定表的副本和磁區規則
    LinkedList<String> list = new LinkedList<String>();
    list.add("CompanyId");
    //設定表副本數
    tableOptions.setNumReplicas(1);
    //設定range磁區
    //tableOptions.setRangePartitionColumns(list);
    //設定hash磁區和磁區的數量
    tableOptions.addHashPartitions(list, 3);
    try {
        kuduClient.createTable("person", schema, tableOptions);
    } catch (Exception e) {
        e.printStackTrace();
    }
    kuduClient.close();
}

六、???????插入資料

/**
 * 向表中加載資料
 * @throws KuduException
 */
@Test
public void loadData() throws KuduException {
    //打開表
    KuduTable kuduTable = kuduClient.openTable(tableName);
    //創建KuduSession物件 kudu必須通過KuduSession寫入資料
    KuduSession kuduSession = kuduClient.newSession();
    //采用flush方式 手動重繪
    kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
    kuduSession.setMutationBufferSpace(3000);
    //準備資料
    for(int i=1; i<=10; i++){
        Insert insert = kuduTable.newInsert();
        //設定欄位的內容
        insert.getRow().addInt("CompanyId",i);
        insert.getRow().addInt("WorkId",i);
        insert.getRow().addString("Name","lisi"+i);
        insert.getRow().addString("Gender","male");
        insert.getRow().addString("Photo","person"+i);
        kuduSession.flush();
        kuduSession.apply(insert);
    }
    kuduSession.close();
    kuduClient.close();
}

七、???????查詢資料

 /**
 * 查詢表資料
 * @throws KuduException
 */
@Test
public void queryData() throws KuduException {
    //打開表
    KuduTable kuduTable = kuduClient.openTable(tableName);
    //獲取scanner掃描器
    KuduScanner.KuduScannerBuilder scannerBuilder = kuduClient.newScannerBuilder(kuduTable);
    KuduScanner scanner = scannerBuilder.build();
    //遍歷
    while(scanner.hasMoreRows()){
        RowResultIterator rowResults = scanner.nextRows();
        while (rowResults.hasNext()){
            RowResult result = rowResults.next();
            int companyId = result.getInt("CompanyId");
            int workId = result.getInt("WorkId");
            String name = result.getString("Name");
            String gender = result.getString("Gender");
            String photo = result.getString("Photo");
            System.out.print("companyId:"+companyId+" ");
            System.out.print("workId:"+workId+" ");
            System.out.print("name:"+name+" ");
            System.out.print("gender:"+gender+" ");
            System.out.println("photo:"+photo);
        }
    }
    //關閉
    scanner.close();
    kuduClient.close();
}

八、修改資料

/**
 * 修改資料
 * @throws KuduException
 */
@Test
public void upDATEData() throws KuduException {
    //打開表
    KuduTable kuduTable = kuduClient.openTable(tableName);
    //構建kuduSession物件
    KuduSession kuduSession = kuduClient.newSession();
    //設定重繪資料模式，自動提交
    kuduSession.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);

    //更新資料需要獲取UpDATE物件
    UpDATE upDATE = kuduTable.newUpDATE();
    //獲取row物件
    PartialRow row = upDATE.getRow();
    //設定要更新的資料資訊
    row.addInt("CompanyId",1);
    row.addString("Name","kobe");
    //操作這個upDATE物件
    kuduSession.apply(upDATE);
    kuduSession.close();
}

九、???????洗掉資料

/**
 * 洗掉表中的資料
 */
@Test
public void deleteData() throws KuduException {
    //打開表
    KuduTable kuduTable = kuduClient.openTable(tableName);
    KuduSession kuduSession = kuduClient.newSession();
    //獲取Delete物件
    Delete delete = kuduTable.newDelete();
    //構建要洗掉的行物件
    PartialRow row = delete.getRow();
    //設定洗掉資料的條件
    row.addInt("CompanyId",2);
    kuduSession.flush();
    kuduSession.apply(delete);
    kuduSession.close();
    kuduClient.close();
}

十、???????修改表

package cn.it.kudu;

import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import org.junit.Before;
import org.junit.Test;

import java.util.List;

/**
 * 修改表操作
 */
public class AlterTable {
    //定義kudu的客戶端物件
    private static KuduClient kuduClient;
    //定義一張表名稱
    private static String tableName = "person";

    /**
     * 初始化操作
     */
    @Before
    public void init() {
        //指定kudu的master地址
        String masterAddress = "node2.cn";
        //創建kudu的資料庫連接
        kuduClient = new KuduClient.KuduClientBuilder(masterAddress).defaultSocketReadTimeoutMs(6000).build();
    }

    /**
     * 添加列
     */
    @Test
    public void alterTableAddColumn() {
        AlterTableOptions alterTableOptions = new AlterTableOptions();
        alterTableOptions.addColumn(new ColumnSchema.ColumnSchemaBuilder("Address", Type.STRING).nullable(true).build());
        try {
            kuduClient.alterTable(tableName, alterTableOptions);
        } catch (KuduException e) {
            e.printStackTrace();
       }
    }

    /**
     * 洗掉列
     */
    @Test
    public void alterTableDeleteColumn(){
        AlterTableOptions alterTableOptions = new AlterTableOptions().dropColumn("Address");
        try {
            kuduClient.alterTable(tableName, alterTableOptions);
        } catch (KuduException e) {
            e.printStackTrace();
       }
    }

    /**
     * 添加磁區列
     */
    @Test
    public void alterTableAddRangePartition(){
        int lowerValue = 110;
        int upperValue = 120;
        try {
            KuduTable kuduTable = kuduClient.openTable(tableName);
            List<Partition> rangePartitions = kuduTable.getRangePartitions(6000);
            boolean flag = true;
            for (Partition rangePartition : rangePartitions) {
                int startKey = rangePartition.getDecodedRangeKeyStart(kuduTable).getInt("Id");
                if(startKey == lowerValue){
                    flag = false;
                }
            }
            if(flag) {
                PartialRow lower = kuduTable.getSchema().newPartialRow();
                lower.addInt("Id", lowerValue);
                PartialRow upper = kuduTable.getSchema().newPartialRow();
                upper.addInt("Id", upperValue);
                kuduClient.alterTable(tableName,new AlterTableOptions().addRangePartition(lower, upper));
            }else{
                System.out.println("磁區已經存在，不能重復創建！");
            }
        } catch (KuduException e) {
            e.printStackTrace();
        } catch (Exception exception) {
            exception.printStackTrace();
        }
    }

    /**
     * 洗掉表
     * @throws KuduException
     */
    @Test
    public void dropTable() throws KuduException {
        kuduClient.deleteTable(tableName);
    }
}

十一、???????洗掉表

/**
 * 洗掉表
 */
@Test
public void dropTable() throws KuduException {
    //洗掉表
    DeleteTableResponse response = kuduClient.deleteTable(tableName);
    //關閉客戶端連接
    kuduClient.close();
}

📢博客主頁：https://lansonli.blog.csdn.net
📢歡迎點贊 👍 收藏 ?留言 📝 如有錯誤敬請指正！
📢本文由 Lansonli 原創，首發于 CSDN博客🙉
📢大資料系列文章會每天更新，停下休息的時候不要忘了別人還在奔跑，希望大家抓緊時間學習，全力奔赴更美好的生活?

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/428564.html

標籤：其他

上一篇：JDK的安裝和卸載

下一篇：Pinpoint【環境搭建 02】Pinpoint Agent 安裝啟動及監控 SpringBoot 專案案例分享

客快物流大資料專案(四十二)：Java代碼操作Kudu

Java代碼操作Kudu

一、???????構建maven工程

二、匯入依賴

三、???????創建包結構

四、???????初始化方法

五、???????創建表

六、???????插入資料

七、???????查詢資料

八、修改資料

九、???????洗掉資料

十、???????修改表

十一、???????洗掉表