| package org.apache.rya.indexing.pcj.fluo.integration; |
| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| import static java.util.Objects.requireNonNull; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.stream.Collectors; |
| |
| import org.apache.fluo.api.client.FluoClient; |
| import org.apache.fluo.api.config.ObserverSpecification; |
| import org.apache.fluo.core.client.FluoClientImpl; |
| import org.apache.kafka.clients.consumer.ConsumerConfig; |
| import org.apache.kafka.clients.consumer.ConsumerRecord; |
| import org.apache.kafka.clients.consumer.ConsumerRecords; |
| import org.apache.kafka.clients.consumer.KafkaConsumer; |
| import org.apache.kafka.common.serialization.StringDeserializer; |
| import org.apache.rya.accumulo.AccumuloRyaDAO; |
| import org.apache.rya.api.domain.RyaStatement; |
| import org.apache.rya.api.domain.RyaSubGraph; |
| import org.apache.rya.api.domain.RyaType; |
| import org.apache.rya.api.domain.RyaURI; |
| import org.apache.rya.api.resolver.RdfToRyaConversions; |
| import org.apache.rya.indexing.pcj.fluo.ConstructGraphTestUtils; |
| import org.apache.rya.indexing.pcj.fluo.api.CreateFluoPcj; |
| import org.apache.rya.indexing.pcj.fluo.app.export.kafka.KafkaSubGraphExporterParameters; |
| import org.apache.rya.indexing.pcj.fluo.app.export.kafka.RyaSubGraphKafkaSerDe; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.AggregationObserver; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.ConstructQueryResultObserver; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.FilterObserver; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.JoinObserver; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.ProjectionObserver; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.QueryResultObserver; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.StatementPatternObserver; |
| import org.apache.rya.indexing.pcj.fluo.app.observers.TripleObserver; |
| import org.apache.rya.pcj.fluo.test.base.KafkaExportITBase; |
| import org.eclipse.rdf4j.model.Statement; |
| import org.eclipse.rdf4j.model.ValueFactory; |
| import org.eclipse.rdf4j.model.impl.SimpleValueFactory; |
| import org.eclipse.rdf4j.model.vocabulary.RDF; |
| import org.eclipse.rdf4j.model.vocabulary.XMLSchema; |
| import org.junit.Assert; |
| import org.junit.Test; |
| |
| import com.google.common.collect.Sets; |
| |
| public class KafkaRyaSubGraphExportIT extends KafkaExportITBase { |
| |
| private static final String BROKERHOST = "127.0.0.1"; |
| private static final String BROKERPORT = "9092"; |
| |
| /** |
| * Add info about the Kafka queue/topic to receive the export. |
| * |
| * @see org.apache.rya.indexing.pcj.fluo.ITBase#setExportParameters(java.util.HashMap) |
| */ |
| @Override |
| protected void preFluoInitHook() throws Exception { |
| // Setup the observers that will be used by the Fluo PCJ Application. |
| final List<ObserverSpecification> observers = new ArrayList<>(); |
| observers.add(new ObserverSpecification(TripleObserver.class.getName())); |
| observers.add(new ObserverSpecification(StatementPatternObserver.class.getName())); |
| observers.add(new ObserverSpecification(JoinObserver.class.getName())); |
| observers.add(new ObserverSpecification(FilterObserver.class.getName())); |
| observers.add(new ObserverSpecification(AggregationObserver.class.getName())); |
| observers.add(new ObserverSpecification(ProjectionObserver.class.getName())); |
| observers.add(new ObserverSpecification(ConstructQueryResultObserver.class.getName())); |
| |
| |
| // Configure the export observer to export new PCJ results to the mini |
| // accumulo cluster. |
| final HashMap<String, String> exportParams = new HashMap<>(); |
| |
| final KafkaSubGraphExporterParameters kafkaParams = new KafkaSubGraphExporterParameters(exportParams); |
| kafkaParams.setUseKafkaSubgraphExporter(true); |
| kafkaParams.setKafkaBootStrapServers(BROKERHOST + ":" + BROKERPORT); |
| |
| final ObserverSpecification exportObserverConfig = new ObserverSpecification(QueryResultObserver.class.getName(), |
| exportParams); |
| observers.add(exportObserverConfig); |
| |
| // Add the observers to the Fluo Configuration. |
| super.getFluoConfiguration().addObservers(observers); |
| } |
| |
| @Test |
| public void basicConstructQuery() throws Exception { |
| // A query that groups what is aggregated by one of the keys. |
| final String sparql = "CONSTRUCT { ?customer <urn:travelsTo> ?city . ?customer <urn:friendsWith> ?worker }" + "WHERE { " |
| + "?customer <urn:talksTo> ?worker. " + "?worker <urn:livesIn> ?city. " + "?worker <urn:worksAt> <urn:burgerShack>. " + "}"; |
| |
| // Create the Statements that will be loaded into Rya. |
| final ValueFactory vf = SimpleValueFactory.getInstance(); |
| final Collection<Statement> statements = Sets.newHashSet( |
| vf.createStatement(vf.createIRI("urn:Joe"), vf.createIRI("urn:talksTo"), vf.createIRI("urn:Bob")), |
| vf.createStatement(vf.createIRI("urn:Bob"), vf.createIRI("urn:livesIn"), vf.createIRI("urn:London")), |
| vf.createStatement(vf.createIRI("urn:Bob"), vf.createIRI("urn:worksAt"), vf.createIRI("urn:burgerShack"))); |
| |
| // Create the PCJ in Fluo and load the statements into Rya. |
| final String pcjId = loadStatements(sparql, statements); |
| |
| // Verify the end results of the query match the expected results. |
| final Set<RyaSubGraph> results = readAllResults(pcjId); |
| |
| final Set<RyaSubGraph> expectedResults = new HashSet<>(); |
| RyaSubGraph subGraph = new RyaSubGraph(pcjId); |
| RyaStatement statement1 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:travelsTo"), new RyaURI("urn:London")); |
| RyaStatement statement2 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:friendsWith"), new RyaURI("urn:Bob")); |
| // if no visibility indicated, then visibilities set to empty byte in |
| // Fluo - they are null by default in RyaStatement |
| // need to set visibility to empty byte so that RyaStatement's equals |
| // will return true |
| statement1.setColumnVisibility(new byte[0]); |
| statement2.setColumnVisibility(new byte[0]); |
| |
| Set<RyaStatement> stmnts = new HashSet<>(Arrays.asList(statement1, statement2)); |
| subGraph.setStatements(stmnts); |
| expectedResults.add(subGraph); |
| |
| ConstructGraphTestUtils.subGraphsEqualIgnoresTimestamp(expectedResults, results); |
| } |
| |
| @Test |
| public void basicConstructQueryWithVis() throws Exception { |
| // A query that groups what is aggregated by one of the keys. |
| final String sparql = "CONSTRUCT { ?customer <urn:travelsTo> ?city . ?customer <urn:friendsWith> ?worker }" + "WHERE { " |
| + "?customer <urn:talksTo> ?worker. " + "?worker <urn:livesIn> ?city. " + "?worker <urn:worksAt> <urn:burgerShack>. " + "}"; |
| |
| // Create the Statements that will be loaded into Rya. |
| RyaStatement statement1 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:talksTo"), new RyaURI("urn:Bob")); |
| RyaStatement statement2 = new RyaStatement(new RyaURI("urn:Bob"), new RyaURI("urn:livesIn"), new RyaURI("urn:London")); |
| RyaStatement statement3 = new RyaStatement(new RyaURI("urn:Bob"), new RyaURI("urn:worksAt"), new RyaURI("urn:burgerShack")); |
| statement1.setColumnVisibility("U&W".getBytes("UTF-8")); |
| statement2.setColumnVisibility("V".getBytes("UTF-8")); |
| statement3.setColumnVisibility("W".getBytes("UTF-8")); |
| |
| // Create the PCJ in Fluo and load the statements into Rya. |
| final String pcjId = loadRyaStatements(sparql, Arrays.asList(statement1, statement2, statement3)); |
| |
| // Verify the end results of the query match the expected results. |
| final Set<RyaSubGraph> results = readAllResults(pcjId); |
| // Create the expected results of the SPARQL query once the PCJ has been |
| // computed. |
| final Set<RyaSubGraph> expectedResults = new HashSet<>(); |
| RyaSubGraph subGraph = new RyaSubGraph(pcjId); |
| RyaStatement statement4 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:travelsTo"), new RyaURI("urn:London")); |
| RyaStatement statement5 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:friendsWith"), new RyaURI("urn:Bob")); |
| // if no visibility indicated, then visibilities set to empty byte in |
| // Fluo - they are null by default in RyaStatement |
| // need to set visibility to empty byte so that RyaStatement's equals |
| // will return true |
| statement4.setColumnVisibility("U&V&W".getBytes("UTF-8")); |
| statement5.setColumnVisibility("U&V&W".getBytes("UTF-8")); |
| |
| Set<RyaStatement> stmnts = new HashSet<>(Arrays.asList(statement4, statement5)); |
| subGraph.setStatements(stmnts); |
| expectedResults.add(subGraph); |
| |
| ConstructGraphTestUtils.subGraphsEqualIgnoresTimestamp(expectedResults, results); |
| } |
| |
| |
| @Test |
| public void constructQueryWithVisAndMultipleSubGraphs() throws Exception { |
| // A query that groups what is aggregated by one of the keys. |
| final String sparql = "CONSTRUCT { ?customer <urn:travelsTo> ?city . ?customer <urn:friendsWith> ?worker }" + "WHERE { " |
| + "?customer <urn:talksTo> ?worker. " + "?worker <urn:livesIn> ?city. " + "?worker <urn:worksAt> <urn:burgerShack>. " + "}"; |
| |
| // Create the Statements that will be loaded into Rya. |
| RyaStatement statement1 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:talksTo"), new RyaURI("urn:Bob")); |
| RyaStatement statement2 = new RyaStatement(new RyaURI("urn:Bob"), new RyaURI("urn:livesIn"), new RyaURI("urn:London")); |
| RyaStatement statement3 = new RyaStatement(new RyaURI("urn:Bob"), new RyaURI("urn:worksAt"), new RyaURI("urn:burgerShack")); |
| RyaStatement statement4 = new RyaStatement(new RyaURI("urn:John"), new RyaURI("urn:talksTo"), new RyaURI("urn:Evan")); |
| RyaStatement statement5 = new RyaStatement(new RyaURI("urn:Evan"), new RyaURI("urn:livesIn"), new RyaURI("urn:SanFrancisco")); |
| RyaStatement statement6 = new RyaStatement(new RyaURI("urn:Evan"), new RyaURI("urn:worksAt"), new RyaURI("urn:burgerShack")); |
| statement1.setColumnVisibility("U&W".getBytes("UTF-8")); |
| statement2.setColumnVisibility("V".getBytes("UTF-8")); |
| statement3.setColumnVisibility("W".getBytes("UTF-8")); |
| statement4.setColumnVisibility("A&B".getBytes("UTF-8")); |
| statement5.setColumnVisibility("B".getBytes("UTF-8")); |
| statement6.setColumnVisibility("C".getBytes("UTF-8")); |
| |
| // Create the PCJ in Fluo and load the statements into Rya. |
| final String pcjId = loadRyaStatements(sparql, Arrays.asList(statement1, statement2, statement3, statement4, statement5, statement6)); |
| |
| // Verify the end results of the query match the expected results. |
| final Set<RyaSubGraph> results = readAllResults(pcjId); |
| // Create the expected results of the SPARQL query once the PCJ has been |
| // computed. |
| RyaStatement statement7 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:travelsTo"), new RyaURI("urn:London")); |
| RyaStatement statement8 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:friendsWith"), new RyaURI("urn:Bob")); |
| RyaStatement statement9 = new RyaStatement(new RyaURI("urn:John"), new RyaURI("urn:travelsTo"), new RyaURI("urn:SanFrancisco")); |
| RyaStatement statement10 = new RyaStatement(new RyaURI("urn:John"), new RyaURI("urn:friendsWith"), new RyaURI("urn:Evan")); |
| statement7.setColumnVisibility("U&V&W".getBytes("UTF-8")); |
| statement8.setColumnVisibility("U&V&W".getBytes("UTF-8")); |
| statement9.setColumnVisibility("A&B&C".getBytes("UTF-8")); |
| statement10.setColumnVisibility("A&B&C".getBytes("UTF-8")); |
| |
| final Set<RyaSubGraph> expectedResults = new HashSet<>(); |
| |
| RyaSubGraph subGraph1 = new RyaSubGraph(pcjId); |
| Set<RyaStatement> stmnts1 = new HashSet<>(Arrays.asList(statement7, statement8)); |
| subGraph1.setStatements(stmnts1); |
| expectedResults.add(subGraph1); |
| |
| RyaSubGraph subGraph2 = new RyaSubGraph(pcjId); |
| Set<RyaStatement> stmnts2 = new HashSet<>(Arrays.asList(statement9, statement10)); |
| subGraph2.setStatements(stmnts2); |
| expectedResults.add(subGraph2); |
| |
| ConstructGraphTestUtils.subGraphsEqualIgnoresTimestamp(expectedResults, results); |
| } |
| |
| @Test |
| public void constructQueryWithBlankNodesAndMultipleSubGraphs() throws Exception { |
| // A query that groups what is aggregated by one of the keys. |
| final String sparql = "CONSTRUCT { _:b <urn:travelsTo> ?city . _:b <urn:friendsWith> ?worker }" + "WHERE { " |
| + "?customer <urn:talksTo> ?worker. " + "?worker <urn:livesIn> ?city. " + "?worker <urn:worksAt> <urn:burgerShack>. " + "}"; |
| |
| // Create the Statements that will be loaded into Rya. |
| RyaStatement statement1 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:talksTo"), new RyaURI("urn:Bob")); |
| RyaStatement statement2 = new RyaStatement(new RyaURI("urn:Bob"), new RyaURI("urn:livesIn"), new RyaURI("urn:London")); |
| RyaStatement statement3 = new RyaStatement(new RyaURI("urn:Bob"), new RyaURI("urn:worksAt"), new RyaURI("urn:burgerShack")); |
| RyaStatement statement4 = new RyaStatement(new RyaURI("urn:John"), new RyaURI("urn:talksTo"), new RyaURI("urn:Evan")); |
| RyaStatement statement5 = new RyaStatement(new RyaURI("urn:Evan"), new RyaURI("urn:livesIn"), new RyaURI("urn:SanFrancisco")); |
| RyaStatement statement6 = new RyaStatement(new RyaURI("urn:Evan"), new RyaURI("urn:worksAt"), new RyaURI("urn:burgerShack")); |
| statement1.setColumnVisibility("U&W".getBytes("UTF-8")); |
| statement2.setColumnVisibility("V".getBytes("UTF-8")); |
| statement3.setColumnVisibility("W".getBytes("UTF-8")); |
| statement4.setColumnVisibility("A&B".getBytes("UTF-8")); |
| statement5.setColumnVisibility("B".getBytes("UTF-8")); |
| statement6.setColumnVisibility("C".getBytes("UTF-8")); |
| |
| // Create the PCJ in Fluo and load the statements into Rya. |
| final String pcjId = loadRyaStatements(sparql, Arrays.asList(statement1, statement2, statement3, statement4, statement5, statement6)); |
| |
| // Verify the end results of the query match the expected results. |
| final Set<RyaSubGraph> results = readAllResults(pcjId); |
| // Create the expected results of the SPARQL query once the PCJ has been |
| // computed. |
| RyaStatement statement7 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:travelsTo"), new RyaURI("urn:London")); |
| RyaStatement statement8 = new RyaStatement(new RyaURI("urn:Joe"), new RyaURI("urn:friendsWith"), new RyaURI("urn:Bob")); |
| RyaStatement statement9 = new RyaStatement(new RyaURI("urn:John"), new RyaURI("urn:travelsTo"), new RyaURI("urn:SanFrancisco")); |
| RyaStatement statement10 = new RyaStatement(new RyaURI("urn:John"), new RyaURI("urn:friendsWith"), new RyaURI("urn:Evan")); |
| statement7.setColumnVisibility("U&V&W".getBytes("UTF-8")); |
| statement8.setColumnVisibility("U&V&W".getBytes("UTF-8")); |
| statement9.setColumnVisibility("A&B&C".getBytes("UTF-8")); |
| statement10.setColumnVisibility("A&B&C".getBytes("UTF-8")); |
| |
| final Set<RyaSubGraph> expectedResults = new HashSet<>(); |
| |
| RyaSubGraph subGraph1 = new RyaSubGraph(pcjId); |
| Set<RyaStatement> stmnts1 = new HashSet<>(Arrays.asList(statement7, statement8)); |
| subGraph1.setStatements(stmnts1); |
| expectedResults.add(subGraph1); |
| |
| RyaSubGraph subGraph2 = new RyaSubGraph(pcjId); |
| Set<RyaStatement> stmnts2 = new HashSet<>(Arrays.asList(statement9, statement10)); |
| subGraph2.setStatements(stmnts2); |
| expectedResults.add(subGraph2); |
| |
| ConstructGraphTestUtils.subGraphsEqualIgnoresBlankNode(expectedResults, results); |
| } |
| |
| |
| @Test |
| public void nestedConstructQuery() throws Exception { |
| // A query that groups what is aggregated by one of the keys. |
| final String sparql = "CONSTRUCT { " |
| + "_:b a <urn:highSpeedTrafficArea> . " |
| + "_:b <urn:hasCount> ?obsCount . " |
| + "_:b <urn:hasLocation> ?location ." |
| + "_:b <urn:hasAverageVelocity> ?avgVelocity ." |
| + "} WHERE { " |
| + "FILTER(?obsCount > 1) " |
| + "{ " |
| + "SELECT ?location (count(?obs) AS ?obsCount) (avg(?velocity) AS ?avgVelocity) " |
| + "WHERE { " |
| + "FILTER(?velocity > 75) " |
| + "?obs <urn:hasVelocity> ?velocity. " |
| + "?obs <urn:hasLocation> ?location. " |
| + "}GROUP BY ?location }}"; |
| |
| // Create the Statements that will be loaded into Rya. |
| final ValueFactory vf = SimpleValueFactory.getInstance(); |
| final Collection<Statement> statements = Sets.newHashSet( |
| vf.createStatement(vf.createIRI("urn:obs1"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(77)), |
| vf.createStatement(vf.createIRI("urn:obs1"), vf.createIRI("urn:hasLocation"), vf.createLiteral("OldTown")), |
| vf.createStatement(vf.createIRI("urn:obs2"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(81)), |
| vf.createStatement(vf.createIRI("urn:obs2"), vf.createIRI("urn:hasLocation"), vf.createLiteral("OldTown")), |
| vf.createStatement(vf.createIRI("urn:obs3"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(70)), |
| vf.createStatement(vf.createIRI("urn:obs3"), vf.createIRI("urn:hasLocation"), vf.createLiteral("OldTown")), |
| vf.createStatement(vf.createIRI("urn:obs5"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(87)), |
| vf.createStatement(vf.createIRI("urn:obs5"), vf.createIRI("urn:hasLocation"), vf.createLiteral("Rosslyn")), |
| vf.createStatement(vf.createIRI("urn:obs6"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(81)), |
| vf.createStatement(vf.createIRI("urn:obs6"), vf.createIRI("urn:hasLocation"), vf.createLiteral("Rosslyn")), |
| vf.createStatement(vf.createIRI("urn:obs7"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(67)), |
| vf.createStatement(vf.createIRI("urn:obs7"), vf.createIRI("urn:hasLocation"), vf.createLiteral("Clarendon")), |
| vf.createStatement(vf.createIRI("urn:obs8"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(77)), |
| vf.createStatement(vf.createIRI("urn:obs8"), vf.createIRI("urn:hasLocation"), vf.createLiteral("Ballston")), |
| vf.createStatement(vf.createIRI("urn:obs9"), vf.createIRI("urn:hasVelocity"), vf.createLiteral(87)), |
| vf.createStatement(vf.createIRI("urn:obs9"), vf.createIRI("urn:hasLocation"), vf.createLiteral("FallsChurch"))); |
| |
| // Create the PCJ in Fluo and load the statements into Rya. |
| final String pcjId = loadStatements(sparql, statements); |
| |
| // Verify the end results of the query match the expected results. |
| final Set<RyaSubGraph> results = readAllResults(pcjId); |
| |
| RyaStatement statement1 = new RyaStatement(new RyaURI("urn:obs1"), new RyaURI("urn:hasCount"), new RyaType(XMLSchema.INTEGER, "2")); |
| RyaStatement statement2 = new RyaStatement(new RyaURI("urn:obs1"), new RyaURI("urn:hasAverageVelocity"), new RyaType(XMLSchema.DECIMAL, "84")); |
| RyaStatement statement3 = new RyaStatement(new RyaURI("urn:obs1"), new RyaURI("urn:hasLocation"), new RyaType("Rosslyn")); |
| RyaStatement statement4 = new RyaStatement(new RyaURI("urn:obs1"), new RyaURI(RDF.TYPE.toString()), new RyaURI("urn:highSpeedTrafficArea")); |
| RyaStatement statement5 = new RyaStatement(new RyaURI("urn:obs2"), new RyaURI("urn:hasCount"), new RyaType(XMLSchema.INTEGER, "2")); |
| RyaStatement statement6 = new RyaStatement(new RyaURI("urn:obs2"), new RyaURI("urn:hasAverageVelocity"), new RyaType(XMLSchema.DECIMAL, "79")); |
| RyaStatement statement7 = new RyaStatement(new RyaURI("urn:obs2"), new RyaURI("urn:hasLocation"), new RyaType("OldTown")); |
| RyaStatement statement8 = new RyaStatement(new RyaURI("urn:obs2"), new RyaURI(RDF.TYPE.toString()), new RyaURI("urn:highSpeedTrafficArea")); |
| |
| final Set<RyaSubGraph> expectedResults = new HashSet<>(); |
| |
| RyaSubGraph subGraph1 = new RyaSubGraph(pcjId); |
| Set<RyaStatement> stmnts1 = new HashSet<>(Arrays.asList(statement1, statement2, statement3, statement4)); |
| subGraph1.setStatements(stmnts1); |
| expectedResults.add(subGraph1); |
| |
| RyaSubGraph subGraph2 = new RyaSubGraph(pcjId); |
| Set<RyaStatement> stmnts2 = new HashSet<>(Arrays.asList(statement5, statement6, statement7, statement8)); |
| subGraph2.setStatements(stmnts2); |
| expectedResults.add(subGraph2); |
| |
| Assert.assertEquals(expectedResults.size(), results.size()); |
| ConstructGraphTestUtils.subGraphsEqualIgnoresBlankNode(expectedResults, results); |
| } |
| |
| |
| protected KafkaConsumer<String, RyaSubGraph> makeRyaSubGraphConsumer(final String TopicName) { |
| // setup consumer |
| final Properties consumerProps = new Properties(); |
| consumerProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKERHOST + ":" + BROKERPORT); |
| consumerProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group0"); |
| consumerProps.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "consumer0"); |
| consumerProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); |
| consumerProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, RyaSubGraphKafkaSerDe.class.getName()); |
| |
| // to make sure the consumer starts from the beginning of the topic |
| consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); |
| |
| final KafkaConsumer<String, RyaSubGraph> consumer = new KafkaConsumer<>(consumerProps); |
| consumer.subscribe(Arrays.asList(TopicName)); |
| return consumer; |
| } |
| |
| private Set<RyaSubGraph> readAllResults(final String pcjId) throws Exception { |
| requireNonNull(pcjId); |
| |
| // Read all of the results from the Kafka topic. |
| final Set<RyaSubGraph> results = new HashSet<>(); |
| |
| try (final KafkaConsumer<String, RyaSubGraph> consumer = makeRyaSubGraphConsumer(pcjId)) { |
| final ConsumerRecords<String, RyaSubGraph> records = consumer.poll(5000); |
| final Iterator<ConsumerRecord<String, RyaSubGraph>> recordIterator = records.iterator(); |
| while (recordIterator.hasNext()) { |
| results.add(recordIterator.next().value()); |
| } |
| } |
| |
| return results; |
| } |
| |
| protected String loadStatements(final String sparql, final Collection<Statement> statements) throws Exception { |
| return loadRyaStatements(sparql, statements.stream().map(x -> RdfToRyaConversions.convertStatement(x)).collect(Collectors.toSet())); |
| } |
| |
| |
| protected String loadRyaStatements(final String sparql, final Collection<RyaStatement> statements) throws Exception { |
| requireNonNull(sparql); |
| requireNonNull(statements); |
| FluoClient client = null; |
| |
| try { |
| CreateFluoPcj createPcj = new CreateFluoPcj(); |
| client = new FluoClientImpl(super.getFluoConfiguration()); |
| String id = createPcj.createPcj(sparql, client).getQueryId(); |
| |
| AccumuloRyaDAO dao = getRyaDAO(); |
| dao.add(statements.iterator()); |
| |
| // Wait for the Fluo application to finish computing the end result. |
| super.getMiniFluo().waitForObservers(); |
| |
| // FluoITHelper.printFluoTable(client); |
| return id; |
| } finally { |
| if (client != null) { |
| client.close(); |
| } |
| } |
| } |
| |
| } |
