使用maven导入kudu
<dependency>
<groupId>org.apache.kudu</groupId>
<artifactId>kudu-client</artifactId>
<version>1.6.</version>
</dependency>
使用java创建Kudu表
public class CreateTable {
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) throws KuduException {
// master地址
final String masteraddr = "hadoop01,hadoop02,hadoop03";
// 创建kudu的数据库链接
KuduClient client = new KuduClient.KuduClientBuilder(masteraddr).defaultSocketReadTimeoutMs(6000).build();
// 设置表的schema
List<ColumnSchema> columns = new LinkedList<ColumnSchema>();
/**
与 RDBMS 不同，Kudu 不提供自动递增列功能，因此应用程序必须始终在插入期间提供完整的主键
*/
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 options = new CreateTableOptions();
// 设置表的replica备份和分区规则
List<String> parcols = new LinkedList<String>();
parcols.add("CompanyId");
//设置表的备份数
options.setNumReplicas(1);
//设置range分区
options.setRangePartitionColumns(parcols);
//设置hash分区和数量
options.addHashPartitions(parcols, 3);
try {
client.createTable("PERSON", schema, options);
} catch (KuduException e) {
e.printStackTrace();
}
client.close();
}
}
使用java删除Kudu表
public class DropTable {
public static void main(String[] args) throws KuduException {
// master地址
final String masteraddr = "hadoop01,hadoop02,hadoop03";
// 创建kudu的数据库链接
KuduClient client = new KuduClient.KuduClientBuilder(masteraddr).defaultSocketReadTimeoutMs(6000).build();
client.deleteTable("PERSON");
client.close();
}
}
使用java插入数据到Kudu
public class InsertRow {
public static void main(String[] args) throws KuduException {
// master地址
final String masteraddr = "hadoop01,hadoop02,hadoop03";
// 创建kudu的数据库链接
KuduClient client = new KuduClient.KuduClientBuilder(masteraddr).build();
// 打开表
KuduTable table = client.openTable("PERSON");
// 创建写session,kudu必须通过session写入
KuduSession session = client.newSession();
// 采取Flush方式 手动刷新
session.setFlushMode(FlushMode.MANUAL_FLUSH);
session.setMutationBufferSpace(3000);
for (int i = 1; i < 10; i++) {
Insert insert = table.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);
session.flush();
session.apply(insert);
}
session.close();
client.close();
}
}
使用java查询Kudu中的数据
public class FindRow {
public static void main(String[] args) throws KuduException {
//master地址
final String masteraddr = "hadoop01,hadoop02,hadoop03";
//创建kudu的数据库链接
KuduClient client = new KuduClient.KuduClientBuilder(masteraddr).defaultSocketReadTimeoutMs(6000).build();
//打开表
KuduTable table = client.openTable("PERSON");
KuduScannerBuilder builder = client.newScannerBuilder(table);
/**
* 设置搜索的条件
* 如果不设置，则全表扫描
*/
KuduPredicate predicate = KuduPredicate.newComparisonPredicate(table.getSchema().getColumn("CompanyId"),
KuduPredicate.ComparisonOp.EQUAL, 1);
builder.addPredicate(predicate);
// 开始扫描
KuduScanner scaner = builder.build();
while (scaner.hasMoreRows()) {
RowResultIterator iterator = scaner.nextRows();
while (iterator.hasNext()) {
RowResult result = iterator.next();
/**
* 输出行
*/
System.out.print("CompanyId:" + result.getInt("CompanyId") +" ");
System.out.print("Name:" + result.getString("Name") +" ");
System.out.print("Gender:" + result.getString("Gender")+" ");
System.out.print("WorkId:" + result.getInt("WorkId") +" ");
System.out.println("Photo:" + result.getString("Photo")+" ");
}
}
scaner.close();
client.close();
}
}
使用java更改Kudu表数据
public class UpdateTable {
public static void main(String[] args) {
// master地址
final String masteraddr = "hadoop01,hadoop02,hadoop03";
// 创建kudu的数据库链接
KuduClient client = new KuduClient.KuduClientBuilder(masteraddr).build();
// 打开表
KuduSession session = null;
try {
KuduTable table = client.openTable("PERSON");
session = client.newSession();
session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
//更新数据
Update update = table.newUpdate();
PartialRow row1 = update.getRow();
row1.addInt("CompanyId",1);
row1.addString("Name","di");
session.apply(update);
} catch (KuduException e) {
e.printStackTrace();
}finally {
try {
session.close();
} catch (KuduException e) {
e.printStackTrace();
}
}
}
}
使用java删除Kudu中指定行数据
public class DeleteRow {
public static void main(String[] args) throws KuduException {
// master地址
final String masteraddr = "hadoop01,hadoop02,hadoop03";
// 创建kudu的数据库链接
KuduClient client = new KuduClient.KuduClientBuilder(masteraddr).build();
// 打开表
KuduTable table = client.openTable("PERSON");
// 创建写session,kudu必须通过session写入
KuduSession session = client.newSession();
final Delete delete = table.newDelete();
//TODO 注意：行删除和更新操作必须指定要更改的行的完整主键;
delete.getRow().addInt("CompanyId" , 5);
session.flush();
session.apply(delete);
session.close();
client.close();
}
}