使用自定義詞彙將結果集作爲RDF流式傳輸

Question

我需要將從查詢中獲得的Jena com.hp.hpl.jena.query.ResultSet流式傳輸到轉換爲RDF輸出格式的遠程端點。 我知道耶拿提供了一個ResultSetFormatter.toModel設施，但是，我有以下限制：

• 我想用一個不同的表示/詞彙，而不是由耶拿提供的一個，和
• 我不想將數據加載到內存中。換句話說，我不想創建一個Model並用ResultSet填充它，但是在我迭代時流出三元組來控制內存消耗。
• 我還是希望由耶拿串行

我已經看到了StreamRDF接口受益，但我不是關於如何有效地使用它非常清楚。 在這種情況下可能是一種正確的方法？

Answer 1

我實施的解決方案適用於我，因此我將其作爲答案發布。 下面的代碼片段應該足夠豐富，因爲它涵蓋了問題中的要求。值得注意的是，我創建了兩個類，一個是ResultSetTripleIterator和QuerySolutionToTripleAdapter接口。第一個負責將三元組傳送給作者，而第二個負責構建來自每個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(); 

然而，耶拿似乎並不不支持某些RDF序列化的流，即XML和JSON格式，導致org.apache.jena.riot.RiotException: No serialization for language Lang:rdf/null，例如。

編輯

由於在耶拿用戶郵件列表反饋，上面的代碼可以利用耶拿的實用工具類，操縱迭代器和執行變換被壓縮。

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