java操作kudu数据库

古城微笑少年丶 2022-05-17 12:22 313阅读 0赞

## ##

### 建表语句： ###

create table user_portrait_kudu(id STRING, open_id STRING, user_id STRING,label_code STRING, label_level SMALLINT, label_name STRING, label_str_value STRING, label_num_value DOUBLE, PRIMARY KEY(id)) PARTITION BY HASH PARTITIONS 16 STORED AS KUDU

### kudu-client代码 ###

<dependency>
       <groupId>org.apache.kudu</groupId>
       <artifactId>kudu-client</artifactId>
       <version>1.2.0</version>
    </dependency>

public interface KuduService {
        /**
         * kudu插入
         * @param data
         * @throws KuduException
         */
        void doInsertProcess(String data) throws KuduException;
    }
    public class KuduServiceImpl implements KuduService, Serializable {
        @Override
        public void doInsertProcess(String data) throws KuduException {
    
            LabelInfo labelInfo = JSON.parseObject(data, LabelInfo.class);
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("id",labelInfo.getId());
            jsonObject.put("label_code",labelInfo.getLabelCode());
            jsonObject.put("label_level",labelInfo.getLabelLevel());
            jsonObject.put("label_name",labelInfo.getLabelName());
            jsonObject.put("label_num_value",labelInfo.getLabelNumValue());
            jsonObject.put("open_id",labelInfo.getOpenId());
            jsonObject.put("user_id",labelInfo.getUserId());
            jsonObject.put("label_str_value",labelInfo.getLabelStrValue());
            KuduUtils.insert("impala::user.user_portrait_kudu", jsonObject);
        }
    }
    public class KuduUtils {
        private static final ThreadLocal<KuduSession> threadLocal = new ThreadLocal<>();
    
        public static KuduTable table(String name) throws KuduException {
            return Kudu.INSTANCE.table(name);
        }
    
        public static Insert emptyInsert(String table) throws KuduException {
            KuduTable ktable = Kudu.INSTANCE.table(table);
            return ktable.newInsert();
        }
    
        public static Update emptyUpdate(String table) throws KuduException {
            KuduTable ktable = Kudu.INSTANCE.table(table);
            return ktable.newUpdate();
        }
    
        public static Upsert emptyUpsert(String table) throws KuduException {
            KuduTable ktable = Kudu.INSTANCE.table(table);
            return ktable.newUpsert();
        }
    
        /**
         * Only columns which are part of the key can be set
         *
         * @param table
         * @return
         */
        public static Delete emptyDelete(String table) throws KuduException {
            KuduTable ktable = Kudu.INSTANCE.table(table);
            return ktable.newDelete();
        }
    
        public static Upsert createUpsert(String table, JSONObject data) throws KuduException {
            KuduTable ktable = Kudu.INSTANCE.table(table);
            Upsert upsert = ktable.newUpsert();
            PartialRow row = upsert.getRow();
            Schema schema = ktable.getSchema();
            for (String colName : data.keySet()) {
                ColumnSchema colSchema = schema.getColumn(colName);
                fillRow(row, colSchema, data);
            }
            return upsert;
        }
    
        public static Insert createInsert(String table, JSONObject data) throws KuduException {
            KuduTable ktable = Kudu.INSTANCE.table(table);
            Insert insert = ktable.newInsert();
            PartialRow row = insert.getRow();
            Schema schema = ktable.getSchema();
            for (String colName : data.keySet()) {
                ColumnSchema colSchema = schema.getColumn(colName);
                fillRow(row, colSchema, data);
            }
            return insert;
        }
        public static void insert(String table, JSONObject data) throws KuduException {
            Insert insert = createInsert(table, data);
            KuduSession session = getSession();
            session.apply(insert);
            session.flush();
            closeSession();
        }
        private static void fillRow(PartialRow row, ColumnSchema colSchema, JSONObject data) {
            String name = colSchema.getName();
            if (data.get(name) == null) {
                return;
            }
            Type type = colSchema.getType();
            switch (type) {
                case STRING:
                    row.addString(name, data.getString(name));
                    break;
                case INT64:
                case UNIXTIME_MICROS:
                    row.addLong(name, data.getLongValue(name));
                    break;
                case DOUBLE:
                    row.addDouble(name, data.getDoubleValue(name));
                    break;
                case INT32:
                    row.addInt(name, data.getIntValue(name));
                    break;
                case INT16:
                    row.addShort(name, data.getShortValue(name));
                    break;
                case INT8:
                    row.addByte(name, data.getByteValue(name));
                    break;
                case BOOL:
                    row.addBoolean(name, data.getBooleanValue(name));
                    break;
                case BINARY:
                    row.addBinary(name, data.getBytes(name));
                    break;
                case FLOAT:
                    row.addFloat(name, data.getFloatValue(name));
                    break;
                default:
                    break;
            }
        }
    
        public static KuduSession getSession() throws KuduException {
            KuduSession session = threadLocal.get();
            if (session == null) {
                session = Kudu.INSTANCE.newAsyncSession();
                threadLocal.set(session);
            }
            return session;
        }
    
        public static KuduSession getAsyncSession() throws KuduException {
            KuduSession session = threadLocal.get();
            if (session == null) {
                session = Kudu.INSTANCE.newAsyncSession();
                threadLocal.set(session);
            }
            return session;
        }
    
        public static void closeSession() {
            KuduSession session = threadLocal.get();
            threadLocal.set(null);
            Kudu.INSTANCE.closeSession(session);
        }
    
    }
    public enum Kudu {
        INSTANCE;
    
        private Kudu() {
            init();
            addShutdownHook();
        }
    
        private KuduClient client = null;
        private Map<String, KuduTable> tables = new HashMap<>();
        private Logger logger = LoggerFactory.getLogger(Kudu.class);
    
        private void init() {
            client = new KuduClient.KuduClientBuilder("192.168.1.132").defaultOperationTimeoutMs(60000)
                    .defaultSocketReadTimeoutMs(30000).defaultAdminOperationTimeoutMs(60000).build();
        }
    
        private void addShutdownHook() {
            Runtime.getRuntime().addShutdownHook(new Thread() {
                public void run() {
                    if (client != null) {
                        try {
                            client.close();
                        } catch (Exception e) {
                            logger.error("ShutdownHook Close KuduClient Error!", e);
                        }
                    }
                }
            });
        }
    
        public KuduClient client() {
            return client;
        }
    
        public KuduTable table(String name) throws KuduException {
            KuduTable table = tables.get(name);
            if (table == null) {
                table = client.openTable(name);
                tables.put(name, table);
            }
            return table;
        }
    
        /**
         * FlushMode:AUTO_FLUSH_BACKGROUND
         *
         * @return
         * @throws KuduException
         */
        public KuduSession newAsyncSession() throws KuduException {
            KuduSession session = client.newSession();
            session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
            session.setFlushInterval(500);
            session.setMutationBufferSpace(5000);
            return session;
        }
    
        /**
         * FlushMode:AUTO_FLUSH_SYNC
         *
         * @return
         * @throws KuduException
         */
        public KuduSession newSession() throws KuduException {
            KuduSession session = client.newSession();
            session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);
            session.setMutationBufferSpace(5000);
            return session;
        }
    
        public void closeSession(KuduSession session) {
            if (session != null && !session.isClosed()) {
                try {
                    session.close();
                } catch (KuduException e) {
                    logger.error("Close KuduSession Error!", e);
                }
            }
        }
    
        public KuduScanner.KuduScannerBuilder scannerBuilder(String table) {
            return client.newScannerBuilder(tables.get(table));
        }
    }