kudu-datastore使用说明
===================================

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

## 创建数据表

.. code:: java

    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();
        }



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

.. code:: java

    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");

        }


## 获取表结构信息

.. code:: java

    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

.. code:: java

    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）

.. code:: java

    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();
    }



## 查询数据

.. code:: java

    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的外包矩形。但是由于统计一般扫描全表，耗时一般很长。调用这些方法时请谨慎使用


.. code:: java

    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个以上的操作。由于实现比较复杂，并且逻辑也不清晰。


.. code:: java

    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();
    }