java - 使用自定义词汇表将结果集流式传输为 RDF

Question

我需要将从查询中获取的 Jena com.hp.hpl.jena.query.ResultSet 流式传输到转换为 RDF 输出格式的远程端点。我知道 Jena 为此提供了一个 ResultSetFormatter.toModel 工具，但是我有以下限制:

• 我想使用不同的表示/词汇，而不是耶拿提供的表示/词汇，并且
• 我不想将数据加载到内存中。换句话说，我不想创建一个 Model 并用 ResultSet 填充它，而是在迭代它时流出三元组，以控制内存消耗。
• 我仍然希望从 Jena 序列化程序中受益

我见过StreamRDF接口(interface)，但我不太清楚如何有效地使用它。在这种情况下，正确的方法是什么？

Answer 1

我实现的解决方案适合我，所以我将其发布为答案。下面的代码片段应该提供足够的信息，因为它涵盖了问题中的要求。值得注意的是，我创建了两个类，一个 ResultSetTripleIterator 和一个 QuerySolutionToTripleAdapter 接口(interface)。第一个负责将三元组流式传输给编写器，而第二个负责从每个 QuerySolution 构建三元组的迭代器。

public class ResultSetTripleIterator implements Iterator<Triple> {
   private ResultSet rs;
   private QuerySolutionToTripleAdapter ad;
   private Iterator<Triple> it = null;
   public ResultSetTripleIterator(ResultSet resultSet, QuerySolutionToTripleAdapter adapter) {
      this.rs = resultSet;
      this.ad = adapter;
   }
   @Override
   public boolean hasNext() {
      if(it != null && it.hasNext()){
         return true;
      }
      it = null;
      return rs.hasNext();
   }
   @Override
   public Triple next() {
       if(it == null){
           it = ad.adapt(rs.next());
       }
       return it.next();
   }
}
public interface QuerySolutionToTripleAdapter {
    public Iterator<Triple> adapt(QuerySolution qs);
}

以下是应用程序示例:

// Can be any OutputStream
OutputStream os = new ByteArrayOutputStream();
StreamRDF stream = StreamRDFWriter.getWriterStream(os, Lang.TRIG);
QueryExecution qe = QueryExecutionFactory.sparqlService(
        "http://data.open.ac.uk/sparql", "PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> SELECT ?A ?B ?C WHERE {?A a ?B . ?A rdf:type ?C} LIMIT 100");

Iterator<Triple> iter = new ResultSetTripleIterator(qe.execSelect(), new QuerySolutionToTripleAdapter() {
    Integer rowIndex = 0;

    @Override
    public Iterator<Triple> adapt(QuerySolution qs) {
        rowIndex++;
        String ns = "http://www.example.org/test/row#";
        String pns = "http://www.example.org/test/col#";
        Resource subject = ResourceFactory.createResource(ns + Integer.toString(rowIndex));
        Property property;
        List<Triple> list = new ArrayList<Triple>();
        Iterator<String> cn = qs.varNames();
        while (cn.hasNext()) {
            String c = cn.next();
            property = ResourceFactory.createProperty(pns + c);
            list.add(new Triple(subject.asNode(), property.asNode(), qs.get(c).asNode()));
        }
        return list.iterator();
    }
});
stream.start();
StreamOps.sendTriplesToStream(iter, stream);
stream.finish();

但是，Jena 似乎并不支持某些用于流式处理的 RDF 序列化，即 XML 和 JSON 格式，从而导致例如 org.apache.jena.riot.RiotException: No serialization for language Lang:rdf/null。

编辑

感谢 Jena 用户邮件列表中的反馈，可以通过使用 Jena 的实用程序类来操作迭代器并执行转换来压缩上面的代码。

代码片段可以重写如下:

OutputStream os = new ByteArrayOutputStream();
StreamRDF stream = StreamRDFWriter.getWriterStream(os, Lang.RDFTHRIFT);
QueryExecution qe = QueryExecutionFactory.sparqlService(
        "http://data.open.ac.uk/sparql", "PREFIX rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> SELECT ?A ?B ?C WHERE {?A a ?B . ?A rdf:type ?C} LIMIT 100");

Transform<QuerySolution, Iterator<Triple>> m = new Transform<QuerySolution, Iterator<Triple>>() {
    Integer rowIndex = 0;

    @Override
    public Iterator<Triple> convert(QuerySolution qs) {
        rowIndex++;
        String ns = "http://www.example.org/test/row#";
        String pns = "http://www.example.org/test/col#";
        Resource subject = ResourceFactory.createResource(ns + Integer.toString(rowIndex));
        Property property;
        List<Triple> list = new ArrayList<Triple>();
        Iterator<String> cn = qs.varNames();
        while (cn.hasNext()) {
            String c = cn.next();
            property = ResourceFactory.createProperty(pns + c);
            list.add(new Triple(subject.asNode(), property.asNode(), qs.get(c).asNode()));
        }
        return list.iterator();
    }
};
Iterator<Triple> iter = WrappedIterator.createIteratorIterator( Iter.map( qe.execSelect(), m ));
stream.start();
StreamOps.sendTriplesToStream(iter, stream);
stream.finish();