1.什么是kudu
在Kudu出现前,由于传统存储系统的局限性,对于数据的快速输入和分析还没有一个完美的解决方案,要么以缓慢的数据输入为代价实现快速分析,要么以缓慢的分析为代价实现数据快速输入。随着快速输入和分析场景越来越多,传统存储层的局限性越来越明显,Kudu应运而生,它的定位介于HDFS和HBase之间,将低延迟随机访问,逐行插入、更新和快速分析扫描融合到一个存储层中,是一个既支持随机读写又支持OLAP分析的存储引擎
kudu应用场景
-
适用于那些既有随机访问,也有批量数据扫描的复合场景
-
高计算量的场景
-
使用了高性能的存储设备,包括使用更多的内存
-
支持数据更新,避免数据反复迁移
-
支持跨地域的实时数据备份和查询
架构
1、Table
表(Table)是数据库中用来存储数据的对象,是有结构的数据集合。kudu中的表具有schema和全局有序的primary key(主键)。kudu中一个table会被水平分成多个被称之为tablet的片段。
2、Tablet
一个 tablet 是一张 table连续的片段,tablet是kudu表的水平分区,类似于HBase的region。每个tablet存储着一定连续range的数据(key),且tablet两两间的range不会重叠。一张表的所有tablet包含了这张表的所有key空间 tablet 会冗余存储。放置到多个 tablet server上,并且在任何给定的时间点,其中一个副本被认为是leader tablet,其余的被认之为follower tablet。每个tablet都可以进行数据的读请求,但只有Leader tablet负责写数据请求
3、Tablet Server
tablet server负责数据存储,并提供数据读写服务 一个 tablet server 存储了table表的tablet,向kudu client 提供读取数据服务。对于给定的 tablet,一个tablet server 充当 leader,其他 tablet server 充当该 tablet 的 follower 副本 只有 leader服务写请求,然而 leader 或 followers 为每个服务提供读请求 。一个 tablet server 可以服务多个 tablets ,并且一个 tablet 可以被多个 tablet servers 服务着
4、Master Server
集群中负责集群管理、元数据管理等功能
2.环境安装
采用docker-composer安装 * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
# Licensed to the Apache Software Foundation (ASF) under one# or more contributor license agreements. See the NOTICE file# distributed with this work for additional information# regarding copyright ownership. The ASF licenses this file# to you under the Apache License, Version 2.0 (the# "License"); you may not use this file except in compliance# with the License. You may obtain a copy of the License at## http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing,# software distributed under the License is distributed on an# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY# KIND, either express or implied. See the License for the# specific language governing permissions and limitations# under the License.version: "3"services: kudu-master-1: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} ports: - "7051:7051" - "8051:8051" command: ["master"] volumes: - kudu-master-1:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 # TODO: Use `host.docker.internal` instead of KUDU_QUICKSTART_IP when it # works on Linux (https://github.com/docker/for-linux/issues/264) - > MASTER_ARGS=--fs_wal_dir=/var/lib/kudu/master --rpc_bind_addresses=0.0.0.0:7051 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP:?Please set KUDU_QUICKSTART_IP environment variable}:7051 --webserver_port=8051 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8051 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=true
kudu-master-2: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} ports: - "7151:7151" - "8151:8151" command: ["master"] volumes: - kudu-master-2:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 - > MASTER_ARGS=--fs_wal_dir=/var/lib/kudu/master --rpc_bind_addresses=0.0.0.0:7151 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP}:7151 --webserver_port=8151 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8151 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=true kudu-master-3: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} ports: - "7251:7251" - "8251:8251" command: ["master"] volumes: - kudu-master-3:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 - > MASTER_ARGS=--fs_wal_dir=/var/lib/kudu/master --rpc_bind_addresses=0.0.0.0:7251 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP}:7251 --webserver_port=8251 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8251 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=true kudu-tserver-1: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} depends_on: - kudu-master-1 - kudu-master-2 - kudu-master-3 ports: - "7050:7050" - "8050:8050" command: ["tserver"] volumes: - kudu-tserver-1:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 - > TSERVER_ARGS=--fs_wal_dir=/var/lib/kudu/tserver --rpc_bind_addresses=0.0.0.0:7050 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP}:7050 --webserver_port=8050 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8050 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=true kudu-tserver-2: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} depends_on: - kudu-master-1 - kudu-master-2 - kudu-master-3 ports: - "7150:7150" - "8150:8150" command: ["tserver"] volumes: - kudu-tserver-2:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 - > TSERVER_ARGS=--fs_wal_dir=/var/lib/kudu/tserver --rpc_bind_addresses=0.0.0.0:7150 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP}:7150 --webserver_port=8150 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8150 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=true kudu-tserver-3: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} depends_on: - kudu-master-1 - kudu-master-2 - kudu-master-3 ports: - "7250:7250" - "8250:8250" command: ["tserver"] volumes: - kudu-tserver-3:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 - > TSERVER_ARGS=--fs_wal_dir=/var/lib/kudu/tserver --rpc_bind_addresses=0.0.0.0:7250 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP}:7250 --webserver_port=8250 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8250 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=true kudu-tserver-4: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} depends_on: - kudu-master-1 - kudu-master-2 - kudu-master-3 ports: - "7350:7350" - "8350:8350" command: ["tserver"] volumes: - kudu-tserver-4:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 - > TSERVER_ARGS=--fs_wal_dir=/var/lib/kudu/tserver --rpc_bind_addresses=0.0.0.0:7350 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP}:7350 --webserver_port=8350 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8350 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=true kudu-tserver-5: image: apache/kudu:${KUDU_QUICKSTART_VERSION:-latest} depends_on: - kudu-master-1 - kudu-master-2 - kudu-master-3 ports: - "7450:7450" - "8450:8450" command: ["tserver"] volumes: - kudu-tserver-5:/var/lib/kudu environment: - KUDU_MASTERS=kudu-master-1:7051,kudu-master-2:7151,kudu-master-3:7251 - > TSERVER_ARGS=--fs_wal_dir=/var/lib/kudu/tserver --rpc_bind_addresses=0.0.0.0:7450 --rpc_advertised_addresses=${KUDU_QUICKSTART_IP}:7450 --webserver_port=8450 --webserver_advertised_addresses=${KUDU_QUICKSTART_IP}:8450 --webserver_doc_root=/opt/kudu/www --stderrthreshold=0 --use_hybrid_clock=false --unlock_unsafe_flags=truevolumes: kudu-master-1: kudu-master-2: kudu-master-3: kudu-tserver-1: kudu-tserver-2: kudu-tserver-3: kudu-tserver-4: kudu-tserver-5:
windows set env * * *
$env:KUDU_QUICKSTART_VERSION = "1.12.0"$env:KUDU_QUICKSTART_IP= "10.11.68.77"Get-ChildItem Env:
linux set env * *
export KUDU_QUICKSTART_VERSION="1.12.0"export KUDU_QUICKSTART_IP=$(ifconfig | grep "inet " | grep -Fv 127.0.0.1 | awk '{print $2}' | tail -1)
run *
docker-compose -f docker/quickstart.yml up -d
stop *
docker-compose -f docker/quickstart.yml down
访问http://localhost:8051/
3.代码工程
实验目的
实现java对kudu的创建表 ,插入,查询,修改操作
pom.xml
<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <parent> <artifactId>springboot-demo</artifactId> <groupId>com.et</groupId> <version>1.0-SNAPSHOT</version> </parent> <modelVersion>4.0.0</modelVersion>
<artifactId>kudu</artifactId>
<properties> <maven.compiler.source>17</maven.compiler.source> <maven.compiler.target>17</maven.compiler.target> <kudu-version>1.12.0</kudu-version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency>
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-autoconfigure</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.kudu</groupId> <artifactId>kudu-client</artifactId> <version>${kudu-version}</version> </dependency>
<!-- For logging messages. --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.30</version> </dependency>
<dependency> <groupId>org.apache.kudu</groupId> <artifactId>kudu-test-utils</artifactId> <version>${kudu-version}</version> <scope>test</scope> </dependency>
</dependencies>
</project>
example.java
// Licensed to the Apache Software Foundation (ASF) under one// or more contributor license agreements. See the NOTICE file// distributed with this work for additional information// regarding copyright ownership. The ASF licenses this file// to you under the Apache License, Version 2.0 (the// "License"); you may not use this file except in compliance// with the License. You may obtain a copy of the License at//// http://www.apache.org/licenses/LICENSE-2.0//// Unless required by applicable law or agreed to in writing,// software distributed under the License is distributed on an// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY// KIND, either express or implied. See the License for the// specific language governing permissions and limitations// under the License.
package com.et.kudu;
import java.util.ArrayList;import java.util.List;
import org.apache.kudu.ColumnSchema;import org.apache.kudu.Schema;import org.apache.kudu.Type;import org.apache.kudu.client.AlterTableOptions;import org.apache.kudu.client.CreateTableOptions;import org.apache.kudu.client.Insert;import org.apache.kudu.client.KuduClient;import org.apache.kudu.client.KuduException;import org.apache.kudu.client.KuduPredicate;import org.apache.kudu.client.KuduPredicate.ComparisonOp;import org.apache.kudu.client.KuduScanner;import org.apache.kudu.client.KuduSession;import org.apache.kudu.client.KuduTable;import org.apache.kudu.client.PartialRow;import org.apache.kudu.client.RowResult;import org.apache.kudu.client.RowResultIterator;import org.apache.kudu.client.SessionConfiguration.FlushMode;
/* * A simple example of using the synchronous Kudu Java client to * - Create a table. * - Insert rows. * - Alter a table. * - Scan rows. * - Delete a table. */public class Example { private static final Double DEFAULT_DOUBLE = 12.345; private static final String KUDU_MASTERS = System.getProperty("kuduMasters", "localhost:7051");
static void createExampleTable(KuduClient client, String tableName) throws KuduException { // Set up a simple schema. List<ColumnSchema> columns = new ArrayList<>(2); columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.INT32) .key(true) .build()); columns.add(new ColumnSchema.ColumnSchemaBuilder("value", Type.STRING).nullable(true) .build()); Schema schema = new Schema(columns);
// Set up the partition schema, which distributes rows to different tablets by hash. // Kudu also supports partitioning by key range. Hash and range partitioning can be combined. // For more information, see http://kudu.apache.org/docs/schema_design.html. CreateTableOptions cto = new CreateTableOptions(); List<String> hashKeys = new ArrayList<>(1); hashKeys.add("key"); int numBuckets = 8; cto.addHashPartitions(hashKeys, numBuckets);
// Create the table. client.createTable(tableName, schema, cto); System.out.println("Created table " + tableName); }
static void insertRows(KuduClient client, String tableName, int numRows) throws KuduException { // Open the newly-created table and create a KuduSession. KuduTable table = client.openTable(tableName); KuduSession session = client.newSession(); session.setFlushMode(FlushMode.AUTO_FLUSH_BACKGROUND); for (int i = 0; i < numRows; i++) { Insert insert = table.newInsert(); PartialRow row = insert.getRow(); row.addInt("key", i); // Make even-keyed row have a null 'value'. if (i % 2 == 0) { row.setNull("value"); } else { row.addString("value", "value " + i); } session.apply(insert); }
// Call session.close() to end the session and ensure the rows are // flushed and errors are returned. // You can also call session.flush() to do the same without ending the session. // When flushing in AUTO_FLUSH_BACKGROUND mode (the mode recommended // for most workloads, you must check the pending errors as shown below, since // write operations are flushed to Kudu in background threads. session.close(); if (session.countPendingErrors() != 0) { System.out.println("errors inserting rows"); org.apache.kudu.client.RowErrorsAndOverflowStatus roStatus = session.getPendingErrors(); org.apache.kudu.client.RowError[] errs = roStatus.getRowErrors(); int numErrs = Math.min(errs.length, 5); System.out.println("there were errors inserting rows to Kudu"); System.out.println("the first few errors follow:"); for (int i = 0; i < numErrs; i++) { System.out.println(errs[i]); } if (roStatus.isOverflowed()) { System.out.println("error buffer overflowed: some errors were discarded"); } throw new RuntimeException("error inserting rows to Kudu"); } System.out.println("Inserted " + numRows + " rows"); }
static void scanTableAndCheckResults(KuduClient client, String tableName, int numRows) throws KuduException { KuduTable table = client.openTable(tableName); Schema schema = table.getSchema();
// Scan with a predicate on the 'key' column, returning the 'value' and "added" columns. List<String> projectColumns = new ArrayList<>(2); projectColumns.add("key"); projectColumns.add("value"); projectColumns.add("added"); int lowerBound = 0; KuduPredicate lowerPred = KuduPredicate.newComparisonPredicate( schema.getColumn("key"), ComparisonOp.GREATER_EQUAL, lowerBound); int upperBound = numRows / 2; KuduPredicate upperPred = KuduPredicate.newComparisonPredicate( schema.getColumn("key"), ComparisonOp.LESS, upperBound); KuduScanner scanner = client.newScannerBuilder(table) .setProjectedColumnNames(projectColumns) .addPredicate(lowerPred) .addPredicate(upperPred) .build();
// Check the correct number of values and null values are returned, and // that the default value was set for the new column on each row. // Note: scanning a hash-partitioned table will not return results in primary key order. int resultCount = 0; int nullCount = 0; while (scanner.hasMoreRows()) { RowResultIterator results = scanner.nextRows(); while (results.hasNext()) { RowResult result = results.next(); if (result.isNull("value")) { nullCount++; } double added = result.getDouble("added"); if (added != DEFAULT_DOUBLE) { throw new RuntimeException("expected added=" + DEFAULT_DOUBLE + " but got added= " + added); } resultCount++; } } int expectedResultCount = upperBound - lowerBound; if (resultCount != expectedResultCount) { throw new RuntimeException("scan error: expected " + expectedResultCount + " results but got " + resultCount + " results"); } int expectedNullCount = expectedResultCount / 2 + (numRows % 2 == 0 ? 1 : 0); if (nullCount != expectedNullCount) { throw new RuntimeException("scan error: expected " + expectedNullCount + " rows with value=null but found " + nullCount); } System.out.println("Scanned some rows and checked the results"); }
public static void main(String[] args) { System.out.println("-----------------------------------------------"); System.out.println("Will try to connect to Kudu master(s) at " + KUDU_MASTERS); System.out.println("Run with -DkuduMasters=master-0:port,master-1:port,... to override."); System.out.println("-----------------------------------------------"); String tableName = "java_example-" + System.currentTimeMillis(); KuduClient client = new KuduClient.KuduClientBuilder(KUDU_MASTERS).build();
try { createExampleTable(client, tableName);
int numRows = 150; insertRows(client, tableName, numRows);
// Alter the table, adding a column with a default value. // Note: after altering the table, the table needs to be re-opened. AlterTableOptions ato = new AlterTableOptions(); ato.addColumn("added", org.apache.kudu.Type.DOUBLE, DEFAULT_DOUBLE); client.alterTable(tableName, ato); System.out.println("Altered the table");
scanTableAndCheckResults(client, tableName, numRows); } catch (Exception e) { e.printStackTrace(); } finally { try { client.deleteTable(tableName); System.out.println("Deleted the table"); } catch (Exception e) { e.printStackTrace(); } finally { try { client.shutdown(); } catch (Exception e) { e.printStackTrace(); } } } }}
以上只是一些关键代码,所有代码请参见下面代码仓库
代码仓库
- https://github.com/Harries/springboot-demo
4.测试
测试创建表
@Testpublic void testCreateExampleTable() throws KuduException { String tableName = "test_create_example"; Example.createExampleTable(client, tableName); assertTrue(client.tableExists(tableName));}
插入数据
@Testpublic void testInsertRows() throws KuduException { String tableName = "test_create_example"; // Example.insertRows(client,tableName,100); System.out.println(client.getTableStatistics(tableName).getLiveRowCount());}
dashboard可以看到我们创建的表和记录条数
5.引用
-
https://kudu.apache.org/
-
http://www.liuhaihua.cn/archives/710904.html