药品监管系统架构揭秘：海量溯源数据存储与查询 ...

依赖

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>tablestore</artifactId>
    <version>4.11.2</version>
</dependency>

Meta表的创建

对于一些固定且有特殊索引需求的字段，我们在创建Meta表的时候需要单独指定，比如“生产日期”、地理信息、状态数据等。

考虑到后面的扩展需求，我们增加一个扩展字段，“extension”，用于存储未定义的元数据。

以下示例只是给了部分元数据字段，用户可以根据自己的需求设置更多的索引字段。

public void createMetaTable() {
    List<AttributeIndexSchema> index = new ArrayList<AttributeIndexSchema>();
    index.add(new AttributeIndexSchema("produced_date", AttributeIndexSchema.Type.LONG));
    index.add(new AttributeIndexSchema("period_of_validity", AttributeIndexSchema.Type.LONG));
    index.add(new AttributeIndexSchema("loc", AttributeIndexSchema.Type.GEO_POINT));
    index.add(new AttributeIndexSchema("links", AttributeIndexSchema.Type.KEYWORD));
    index.add(new AttributeIndexSchema("status", AttributeIndexSchema.Type.KEYWORD));
    index.add(new AttributeIndexSchema("extension", AttributeIndexSchema.Type.KEYWORD).setIsArray(true));
    db.createMetaTable(index);
}

Data表的创建

这个比较简单，只需要设定表名即可。因为我们是Schema Free的体系，不需要预先指定列，在写入的时候指定即可。

public void createDataTable() {
    db.createDataTable(conf.getDataTableName());
}

录入药品元数据和状态数据

元数据导入，我们将一个本地的csv文件中的数据导入到数据库中

public void importMeta() throws IOException {
    TimestreamMetaTable metaTable = db.metaTable();
    String [] fileHeader = {"分类", "名称", "监管号", "受理号", "生产日期", "有效日期", "注册分类", "申请类型", "企业名称", "任务类型"};
    String csvFile = conf.getMetaFile();
    CSVFormat format = CSVFormat.DEFAULT.withHeader(fileHeader).withIgnoreHeaderCase().withTrim();
    Reader reader = Files.newBufferedReader(Paths.get(csvFile));
    CSVParser csvParser = new CSVParser(reader, format);
    for (CSVRecord r : csvParser.getRecords()) {
        TimestreamIdentifier identifier = new TimestreamIdentifier.Builder(r.get("分类"))
                .addTag("名称", r.get("名称"))
                .addTag("监管号", r.get("监管号"))
                .build();
        TimestreamMeta meta = new TimestreamMeta(identifier);

        meta.addAttribute("produced_date", r.get("生产日期"));
        meta.addAttribute("period_of_validity", r.get("有效日期"));

        List<String> extension = new ArrayList();
        extension.add("受理号=" + r.get("受理号"));
        extension.add("注册分类=" + r.get("注册分类"));
        extension.add("申请类型=" + r.get("申请类型"));
        extension.add("企业名称=" + r.get("企业名称"));
        extension.add("任务类型=" + r.get("任务类型"));
        meta.addAttribute("extension", new Gson().toJson(extension));

        metaTable.put(meta);
        System.out.println(meta.toString());
    }
}

状态数据导入，这里loc, links,status在Meta和Data都存储了一次，Meta表中存储主要是做后续的索引查询，Data表中存储主要是做

public void importData() throws Exception {
    TimestreamMetaTable metaTable = db.metaTable();
    TimestreamDataTable dataTable = db.dataTable(conf.getDataTableName());

    String [] fileHeader = {"分类", "名称", "监管号", "生产日期", "位置", "环节", "状态"};
    String csvFile = conf.getDataFile();
    CSVFormat format = CSVFormat.DEFAULT.withHeader(fileHeader).withIgnoreHeaderCase().withTrim();
    Reader reader = Files.newBufferedReader(Paths.get(csvFile));
    CSVParser csvParser = new CSVParser(reader, format);
    for (CSVRecord r : csvParser.getRecords()) {
        TimestreamIdentifier identifier = new TimestreamIdentifier.Builder(r.get("分类"))
                .addTag("名称", r.get("名称"))
                .addTag("监管号", r.get("监管号"))
                .build();

        TimestreamMeta meta = new TimestreamMeta(identifier);

        String loc = toLocationString(r.get("位置"));
        String links = r.get("环节");
        String status = r.get("状态");

        meta.addAttribute("loc", loc);
        meta.addAttribute("links", links);
        meta.addAttribute("status", status);
        metaTable.update(meta);

        Point point = new Point.Builder(this.getTimestamp(r, "生产日期"), TimeUnit.MILLISECONDS)
                .addField("loc", loc)
                .addField("links", links)
                .addField("status", status)
                .build();
        dataTable.asyncWrite(identifier, point);
        System.out.println(point.toString());
    }
    dataTable.flush();
}

多维度查询药品溯源信息

1. 基本的药品详细信息查询，主要是根据用户输入条件，显示药品的元数据。我们这里根据药品分类、药品名称、生产企业来查询药品。

Filter filter = and(
        Name.equal("中药"),                                    
        Tag.equal("名称", "复方阿胶"),                      
        Attribute.in("extension", new String[]{"企业名称=山东****也有限公司"})
);

Iterator<TimestreamMeta> iter = metaTable.filter(filter).fetchAll();
while (iter.hasNext()) {
    TimestreamMeta m = iter.next();
    System.out.println(m);
}

2. 药品的防伪鉴定，结合生产日期，运输轨迹、销售状态和查询用户等数据对药品实行防伪鉴定。我们这里输入名称和药品监管码。

Filter filter = and(
        Name.equal("中药"),                                 
        Tag.equal("名称", "复方阿胶"),                         
        Tag.equal("监管号", "8160000000000019")               
);
Iterator<TimestreamMeta> iter = metaTable.filter(filter).selectAttributes("status").fetchAll();
while (iter.hasNext()) {
    TimestreamMeta m = iter.next();
    System.out.println(m.getAttributeAsString("status"));
}
// 从查询的结果来看，药品处于召回中，有使用风险

3. 查询指定地点范围内的特定药品。比如查询使用者5KM范围的“阿莫西林”。

Filter filter = and(
        Name.equal("化药"),
        Tag.prefix("名称", "阿莫西林"),
        Attribute.inGeoDistance("loc", "31.6533906593,103.8427768645", 5 * 1000)
);
Iterator<TimestreamMeta> iter = metaTable.filter(filter).fetchAll();
while (iter.hasNext()) {
    TimestreamMeta m = iter.next();
    System.out.println(m);
}

4. 药品轨迹重放，遍历指定药品的一个轨迹溯源信息。

TimestreamIdentifier identifier = new TimestreamIdentifier.Builder("化药")
        .addTag("名称", "阿莫西林")
        .addTag("监管号", "8150000000000000")
        .build();

Iterator<Point> iter = dataTable.get(identifier).select("loc").fetchAll();
while (iter.hasNext()) {
    Point p = iter.next();
    System.out.println(p);
}

相关代码

可以在github上获取实现代码和示例代码，欢迎大家体验、使用以及给我们提出建议。

代码链接： https://github.com/aliyun/tablestore-examples/tree/master/demos/TraceMedicine 

欢迎加入