使用javaAPI操作kudu

1.1、Kudu API的结构设计 1.1.1、Client

1.1.2、Table

1.1.3、Scanner

1.2、导入依赖

导入kudu依赖

        
            org.apache.kudu
            kudu-client
            1.7.1
        

1.3、操作 1.3.1、初始化

  var kuduClient: KuduClient = null;

  // 初始化
  @Before
  def initKuduClient() = {
    // 1、创建Client
    val KUDU_MASTER = "s202:7051,s203:7051,s204:7051"
    kuduClient = new KuduClientBuilder(KUDU_MASTER).build()
  }

1.3.2、创建表

  @Test
  def createTable(): Unit = {
    // 2、创建schema
    val columns = List(
      new ColumnSchema.ColumnSchemaBuilder("id", Type.STRING).key(true).build(),
      new ColumnSchema.ColumnSchemaBuilder("name", Type.STRING).key(false).build()
    )
    // 隐式转换，
    import scala.collection.JavaConverters._
    val schema = new Schema(columns.asJava)

    // 3、创建分区
    val options = new CreateTableOptions()
      .setRangePartitionColumns(List("id").asJava)
      .setNumReplicas(1) // tablet是存储数据的位置，tablet可以有多个，有主从概念

    // 4、创建表
    kuduClient.createTable("kudutest1", schema, options)
  }

1.3.3、插入数据

此处有个地方需要注意，如果在插入一条数相同id的数据，kudu API不会报错

  // 插入数据
  @Test
  def insert(): Unit = {
    // 获取table, 表示对一个表进行操作
    val table = kuduClient.openTable("kudutest")

    // 获取一个操作， 组织对表数据的操作
    val insert = table.newInsert()
    val row = insert.getRow
    row.addString("id","1")
    row.addString("name","chb")

    // 开启会话，执行操作
    val session = kuduClient.newSession()
    session.apply(insert)
  }

1.3.4、更新数据

  // 更新数据
  @Test
  def update(): Unit = {
    // 获取table, 表示对一个表进行操作
    val table = kuduClient.openTable("kudutest")

    // 获取一个操作， 组织对表数据的操作
    val update = table.newUpdate()
    val row = update.getRow
    row.addString("id","1")
    row.addString("name","ling")

    // 开启会话，执行操作
    val session = kuduClient.newSession()
    session.apply(update)
  }

1.3.5、删除数据

  // 删除数据
  @Test
  def delete(): Unit = {
    // 获取table, 表示对一个表进行操作
    val table = kuduClient.openTable("kudutest")

    // 获取一个操作， 组织对表数据的操作
    val delete = table.newDelete()
    val row = delete.getRow
    row.addString("id", "1")


    // 开启会话，执行操作
    val session = kuduClient.newSession()
    session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC)
    session.apply(delete)
  }

1.3.6、查询数据

  @Test
  // 查询
  def query(): Unit = {
    // 1、构建一个扫描器
    var kuduScannerBuilder = kuduClient.newScannerBuilder(kuduClient.openTable("kudutest"))

    // 隐式转化，scala->java
    import scala.collection.JavaConverters._
    val columnNames = List("id", "name").asJava
    kuduScannerBuilder = kuduScannerBuilder.setProjectedColumnNames(columnNames)

    // 结果集
    val scanner = kuduScannerBuilder.build()
    while (scanner.hasMoreRows) {
      // 注意，这rows是获取一整个 tablet的数据
      val rows = scanner.nextRows()

      while (rows.hasNext) {
        val row = rows.next()
        println(row.getString(0), row.getString(1))
      }

    }
  }

代码地址