126.在CDH中用Java API访问Kudu

126.1 准备 环境

Kudu配置修改

# 在KuduMaster服务的高级配置” gflagfile 的 Master 高级配置代码段（安全阀）”增加配置
--trusted_subnets=0.0.0.0/0
# 由于Fayson使用的是AWS环境这里跨了网段需要进行配置，如果局域网可以跳过此步

# 如果未配置在使用Java API访问Kudu时报如下错误
W1128 16:56:55.749083 93981 negotiation.cc:318] Unauthorized connection attempt: Server connection negotiation failed: server connection from 100.73.0.57:42533: unauthenticated connections from publicly routable IPs are prohibited. See --trusted_subnets flag for more information

126.2 Java 示例

Kudu的Maven工程创建
pom.xml 文件内容

<dependencies>
    <dependency>
        <groupId>org.apache.kudu</groupId>
        <artifactId>kudu-client</artifactId>
        <version>1.6.0</version>
    </dependency>
    <!-- For logging messages. -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.25</version>
    </dependency>
    <dependency>
        <groupId>com.cloudera</groupId>
        <artifactId>generate-data</artifactId>
        <version>1.0-SNAPSHOT</version>
    </dependency>
</dependencies>

KuduEx ample.java类

package com.cloudera;
import org.apache.kudu.ColumnSchema;
import org.apache.kudu.Schema;
import org.apache.kudu.Type;
import org.apache.kudu.client.*;
import java.util.ArrayList;
import java.util.List;

public class KuduExample {
    /**
     * 使用Kudu API创建一个Kudu表
     * @param client
     * @param tableName
     */
    public static void createTable(KuduClient client, String tableName) {
        List<ColumnSchema> columns = new ArrayList<>();
        //在添加列时可以指定每一列的压缩格式
        columns.add(new ColumnSchema.ColumnSchemaBuilder("id", Type.STRING).key(true).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("name", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("sex", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("city", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("occupation", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("tel", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("fixPhoneNum", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("bankName", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("address", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("marriage", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        columns.add(new ColumnSchema.ColumnSchemaBuilder("childNum", Type.STRING).
                compressionAlgorithm(ColumnSchema.CompressionAlgorithm.SNAPPY).build());
        Schema schema = new Schema(columns);
        CreateTableOptions createTableOptions = new CreateTableOptions();
        List<String> hashKeys = new ArrayList<>();
        hashKeys.add("id");
        int numBuckets = 8;
        createTableOptions.addHashPartitions(hashKeys, numBuckets);
        try {
            if(!client.tableExists(tableName)) {
                client.createTable(tableName, schema, createTableOptions);
            }
            System.out.println("成功创建Kudu表：" + tableName);
        } catch (KuduException e) {
            e.printStackTrace();
        }
    }
    /**
     * 向指定的Kudu表中upsert数据，数据存在则更新，不存在则新增
     * @param client KuduClient对象
     * @param tableName 表名
     * @param numRows 向表中插入的数据量
     */
    public static void upsert(KuduClient client, String tableName, int numRows ) {
        try {
            KuduTable kuduTable = client.openTable(tableName);
            KuduSession kuduSession = client.newSession();
            //设置Kudu提交数据方式，这里设置的为手动刷新，默认为自动提交
            kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
            for(int i =0; i < numRows; i++) {
                String userInfo_str = RandomUserInfo.getUserInfo("测试数据");
                Upsert upsert = kuduTable.newUpsert();
                PartialRow row = upsert.getRow();
                String[] userInfo = userInfo_str.split(",");
                if(userInfo.length == 11) {
                    row.addString("id", userInfo[0]);
                    row.addString("name", userInfo[1]);
                    row.addString("sex", userInfo[2]);
                    row.addString("city", userInfo[3]);
                    row.addString("occupation", userInfo[4]);
                    row.addString("tel", userInfo[5]);
                    row.addString("fixPhoneNum", userInfo[6]);
                    row.addString("bankName", userInfo[7]);
                    row.addString("address", userInfo[8]);
                    row.addString("marriage", userInfo[9]);
                    row.addString("childNum", userInfo[10]);
                }
                kuduSession.apply(upsert);
            }
            kuduSession.flush();
            kuduSession.close();
        } catch (KuduException e) {
            e.printStackTrace();
        }
    }
    /**
     * 查看Kudu表中数据
     * @param client
     * @param tableName
     */
    public static void scanerTable(KuduClient client, String tableName) {
        try {
            KuduTable kuduTable = client.openTable(tableName);
            KuduScanner kuduScanner = client.newScannerBuilder(kuduTable).build();
            while(kuduScanner.hasMoreRows()) {
                RowResultIterator rowResultIterator =kuduScanner.nextRows();
                while (rowResultIterator.hasNext()) {
                    RowResult rowResult = rowResultIterator.next();
                    System.out.println(rowResult.getString("id"));
                }
            }
            kuduScanner.close();
        } catch (KuduException e) {
            e.printStackTrace();
        }
    }
    /**
     * 删除表
     * @param client
     * @param tableName
     */
    public static void dropTable(KuduClient client, String tableName) {
        try {
            client.deleteTable(tableName); 
        } catch (KuduException e) {
            e.printStackTrace();
        }
    }
    /**
     * 列出Kudu下所有的表
     * @param client
     */
    public static void tableList(KuduClient client) {
        try {
            ListTablesResponse listTablesResponse = client.getTablesList();
            List<String> tblist = listTablesResponse.getTablesList();
            for(String tableName : tblist) {
                System.out.println(tableName);
            }
        } catch (KuduException e) {
            e.printStackTrace();
        }
    }
    private static final String KUDU_MASTER = System.getProperty("kuduMasters", "ip-172-31-5-38.ap-southeast-1.compute.internal:7051,ip-172-31-7-193.ap-southeast-1.compute.internal:7051,ip-172-31-8-230.ap-southeast-1.compute.internal:7051");
    public static void main(String[] args) {
        KuduClient kuduClient = new KuduClient.KuduClientBuilder(KUDU_MASTER).build();
        String tableName = "user_info";
        //删除Kudu的表
        dropTable(kuduClient, tableName);
        //创建一个Kudu的表
        createTable(kuduClient, tableName);
        //列出Kudu下所有的表
        tableList(kuduClient);
        //向Kudu指定的表中插入数据
        upsert(kuduClient, tableName, 100);
        //扫描Kudu表中数据
        scanerTable(kuduClient, tableName);
        try {
            kuduClient.close();
        } catch (KuduException e) {
            e.printStackTrace();
        }
    }
}

运行

删除已存在的user_info表
创建一个新的user_info表
查看Kudu中所有的表名
向user_info中插入100条数据
扫描user_info表数据

126.3 KuduMaster查看 创建的表

126.4 Impala访问 集成

建表语句

CREATE EXTERNAL TABLE `user_info` STORED AS KUDU
TBLPROPERTIES(
    'kudu.table_name' = 'user_info',
    'kudu.master_addresses' = 'ip-xxx-xx-x-xx.ap-southeast-1.compute.internal:7051,ip-xxx-xx-x-xx.ap-southeast-1.compute.internal:7051,ip-xxx-xx-x-xx.ap-southeast-1.compute.internal:7051')

1.在使用Java API访问Kudu时如果跨了网络则需要增加配置--trusted_subnets=0.0.0.0/0将网络添加到受新人列表
2.通过Java API接口创建的Kudu表，默认是没有与Impala集成的，需要通过KuduMaster中提供的SQL在Impala中执行。

原文地址:https://blog.csdn.net/m0_47454596/article/details/127091846

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。

如若转载，请注明出处：http://www.7code.cn/show_22516.html

如若内容造成侵权/违法违规/事实不符，请联系代码007邮箱：suwngjj01@126.com进行投诉反馈，一经查实，立即删除！

声明：本站所有文章，如无特殊说明或标注，均为本站原创发布。任何个人或组织，在未征得本站同意时，禁止复制、盗用、采集、发布本站内容到任何网站、书籍等各类媒体平台。如若本站内容侵犯了原著者的合法权益，可联系我们进行处理。

126.1 准备环境

126.2 Java示例

126.3 KuduMaster查看创建的表

126.4 Impala访问集成

相关文章

发表回复 取消回复

126.2 Java 示例

发表回复取消回复