2.1. kudu-datastore使用说明

## kudu-datastore介绍 kudu-datastore提供符合GeoTools规范的SDK二次开发接口，目的在于使用户可以在分布式kudu数据库中存储和管理海量空间数据。通过该接口实现对时空数据的管理与操作。

## 创建数据表

public static void main(String[] args) throws Exception {
        KuduDataStoreFactory factory = new KuduDataStoreFactory();
        Map<String, Serializable> params = new HashMap<>();
        //kudu主节点IP
        params.put(KuduMasterParam.key, "192.168.20.11:7051");
        //元数据表名称
        params.put(CatalogParam.key, "catalog");
        //设置查询的线程数
        params.put(QueryThreadsParam.key, 5);
        //创建一个kuduDatastore对象
        KuduDataStore dataStore = factory.createDataStore(params);
        //根据shapefile进行表创建，读取shapefile信息
        File file = new File("C:\\Users\\user\\Desktop\\国界\\guojie.shp");

        //查询typeName元数据
        dataStore.metadata().scan("guojie", "")
                .forEach((k, v) -> System.out.println(k + ": " + v));

        //删除typeName
        dataStore.removeSchema("guojie");

        ShapefileDataStore shapeFileDataStore = (ShapefileDataStore) new ShapefileDataStoreFactory().createDataStore(file.toURL());
        SimpleFeatureType sft = shapeFileDataStore.getFeatureReader().getFeatureType();

        //配置索引方式
        sft.getUserData().put(Configs.EnabledIndices, "xz2");

        //设置BBOX
        ReferencedEnvelope env = source.getBounds();
        sft.getUserData().put(RANGE_KEY, env);

        //可以自定义表名
        sft = SimpleFeatureTypes.renameSft(featureType, "自定义表名");
        //创建typeName
        dataStore.createSchema(sft);
        shapeFileDataStore.dispose();
        dataStore.dispose();
    }

## 查询元数据信息和删除表

public static void main(String[] args) throws Exception {
        KuduDataStoreFactory factory = new KuduDataStoreFactory();
        Map<String, Serializable> params = new HashMap<>();
        //kudu主节点IP
        params.put(KuduMasterParam.key, "192.168.20.11:7051");
        //元数据表名称
        params.put(CatalogParam.key, "catalog");
        //设置查询的线程数
        params.put(QueryThreadsParam.key, 5);
        //创建一个kuduDatastore对象
        KuduDataStore dataStore = factory.createDataStore(params);

        //查询typeName元数据
        dataStore.metadata().scan("guojie", "")
                .forEach((k, v) -> System.out.println(k + ": " + v));

        //删除表
        dataStore.removeSchema("guojie");

    }

## 获取表结构信息

public Map getSchema(String tableName) throws IOException {
        Map params = new HashMap();
        params.put("kudu.master", "192.168.3.194:7051");
        params.put("kudu.catalog", "changsha_demo");

        KuduDataStoreFactory factory = new KuduDataStoreFactory();
        KuduDataStore dataStore = (KuduDataStore) factory.createDataStore(params);
        SimpleFeatureType schema = dataStore.getSchema(tableName);
        Map map =new HashMap();
        List<Map> col_list = new ArrayList<Map>();
        List<AttributeDescriptor> attributeDescriptors = schema.getAttributeDescriptors();
        for(AttributeDescriptor attributeDescriptor:attributeDescriptors){
            Map temp = new HashMap();
            temp.put(attributeDescriptor.getType().getName().toString(),attributeDescriptor.getType().getBinding().getSimpleName().toString());
            col_list.add(temp);
        }
        map.put("col_list",col_list);

        return map;
    }

## 将shapeFile中数据插入kudu

public static void main(String[] args) throws IOException {
    KuduDataStoreFactory factory = new KuduDataStoreFactory();
    Map<String, Serializable> params = new HashMap<>();
    params.put(KuduMasterParam.key, "192.168.20.11:7051");
    params.put(CatalogParam.key, "catalog");
    params.put(QueryThreadsParam.key, 5);

    KuduDataStore dataStore = factory.createDataStore(params);

    //shapefile文件地址
    File file = new File("C:\\Users\\user\\Desktop\\国界\\guojie.shp");

    ShapefileDataStore shapeFileDataStore = (ShapefileDataStore) new ShapefileDataStoreFactory().createDataStore(file.toURL());
    FeatureReader<SimpleFeatureType, SimpleFeature> reader = shapeFileDataStore.getFeatureReader();
    SimpleFeatureType sft = reader.getFeatureType();
    int i = 0;

    QueryPlannedFeatureWriter writer = (QueryPlannedFeatureWriter) dataStore.getFeatureWriter("guojie", Filter.EXCLUDE, Transaction.AUTO_COMMIT);

    while (reader.hasNext()) {
        SimpleFeature feature = reader.next();
        SimpleFeature cur = (SimpleFeature) writer.next();

        for (AttributeDescriptor d : sft.getAttributeDescriptors()) {
            cur.setAttribute(d.getName(), feature.getAttribute(d.getName()));
        }

        writer.write();
        i++;
    }

    System.out.println("共插入数据"+i+"条");
    reader.close();
    shapeFileDataStore.dispose();
    dataStore.dispose();
}

## 独立创建属性索引表 当创建几何索引表以后，可以独立的创建属性索引表。如果几何索引表中有数据时，属性索引表创建成功后，所有数据也将插入完成。另外，数据没有插入完成时，同时查询typeName，没有完成的索引表，不会被查询。调用方法dataStore.createAttributeIndex（String typeName, String attr）

public static void main(String[] args) throws Exception {
        KuduDataStoreFactory factory = new KuduDataStoreFactory();
        Map<String, Serializable> params = new HashMap<>();
        params.put(KuduMasterParam.key, "192.168.20.11:7051");
        params.put(CatalogParam.key, "catalog");
        params.put(QueryThreadsParam.key, 5);

        KuduDataStore dataStore = factory.createDataStore(params);

        //创建属性索引：
        dataStore.createAttributeIndex("guojie", "LENGTH");

        dataStore.dispose();
}

## 查询数据

public static void main(String[] args) throws Exception {

        KuduDataStoreFactory factory = new KuduDataStoreFactory();
        Map<String, Serializable> params = new HashMap<>();
        params.put(KuduMasterParam.key, "192.168.20.11:7051");
        params.put(CatalogParam.key, "catalog");
        params.put(QueryThreadsParam.key, 5);

        KuduDataStore dataStore = factory.createDataStore(params);

        SimpleFeatureType sft = dataStore.getSchema("guojie");
        FilterFactory2 ff = CommonFactoryFinder.getFilterFactory2();

        Filter filter1 = ff.intersects(ff.property("the_geom"), ff.literal("MULTILINESTRING ((113.273918 8.111534, 113.270233 8.1111, 113.269073 8.108811, 113.269051 8.107203, 113.27066 8.106962, 113.273453 8.109011, 113.274384 8.109922, 113.273918 8.111534))"));



        Filter all = ff.or(filter, filter7);
        String[] attrs = new String[4];
        attrs[0] = "FNODE_";
        attrs[1] = "TNODE_";
        attrs[2] = "LENGTH";
        attrs[3] = "length(the_geom)";

        long begin = System.currentTimeMillis();
        Query query = new Query("guojie");
        //设置最多返回数据条数
        query.setMaxFeatures(10);

        QueryPlannedFeatureReader reader = (QueryPlannedFeatureReader) dataStore.getFeatureReader(query, Transaction.AUTO_COMMIT);
        long end = System.currentTimeMillis();
        int i = 0;

        while (reader.hasNext()) {
            SimpleFeature sf = reader.next();
            System.out.println(sf);
            i++;
        }
        System.out.println(i);
        System.out.println(end - begin);

        reader.close();
        dataStore.dispose();
    }

## 统计查询 统计查询可以查询某个typeName的的统计信息，例如总记录条数；某个字段的最大最少值；typeName的外包矩形。但是由于统计一般扫描全表，耗时一般很长。调用这些方法时请谨慎使用

public static void main(String[] args) throws IOException {

        KuduDataStoreFactory factory = new KuduDataStoreFactory();
        Map<String, Serializable> params = new HashMap<>();
        params.put(KuduMasterParam.key, "192.168.20.11:7051");
        params.put(CatalogParam.key, "catalog");
        params.put(QueryThreadsParam.key, 5);

        KuduDataStore dataStore = factory.createDataStore(params);

    long begin = System.currentTimeMillis();
    Query query = new Query("T530502_F");

    //必须设置为true,否则返回估计信息或者默认信息
    query.getHints().put(EXACT_COUNT, true);

    GenericFeatureSource source = dataStore.getFeatureSource(query.getTypeName());
    SimpleFeatureIterator iter = source.getFeatures(query).features();

//统计数据的总记录数，比较耗时
    long count = source.getCount(query);
    System.out.println(count);

    //统计数据的外包矩形，非常耗时
    ReferencedEnvelope mm = source.getBounds(query);
    System.out.println(mm.getMinX() + "," + mm.getMaxX() + "," + mm.getMinY() + "," + mm.getMaxY());

    //System.out.println(sum);
    System.out.println(System.currentTimeMillis() - begin);
    iter.close();
    dataStore.dispose();
}

## schema更新 KuduDatastore支持的schema更新方式： Rename typeName; Append property; Drop a propery; Change poperty name

Schema更新的后面三个方式必须是互斥的，即一次只能append, 或者 drop 或者change，不能同时做2个或者2个以上的操作。由于实现比较复杂，并且逻辑也不清晰。

public static void main(String[] args) throws IOException {
        Map params = new HashMap();
        params.put("kudu.master", "192.168.20.11:7051");
        params.put("kudu.catalog", "catalog");
        // 创建kuduDatasotre对象
        KuduDataStoreFactory factory = new KuduDataStoreFactory();
        KuduDataStore dataStore = (KuduDataStore) actory.createDataStore(params);
        Query query = new Query("guojie");
        SimpleFeatureReader featureReader = dataStore.getFeatureReader(query, Transaction.AUTO_COMMIT);
        SimpleFeatureType featureType = featureReader.getFeatureType();

        SimpleFeatureTypeBuilder builder = new SimpleFeatureTypeBuilder();
        for(AttributeDescriptor ad : featureType.getAttributeDescriptors()){
            builder.add(ad.getType().getName().getLocalPart(), ad.getType().getBinding());
        }

        //添加一列：
        builder.add("GBCODE_N", Integer.class);
        builder.setName("gj");

        builder.setDefaultGeometry("the_geom");
        SimpleFeatureType sft = builder.buildFeatureType();
        sft.getUserData().putAll(featureType.getUserData());


        dataStore.updateSchema("guojie", sft);

        dataStore.metadata().scan("guojie")
                .forEach((k, v) -> System.out.println(k + ", " + v));
        dataStore.dispose();
}