import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * Compact all transient regions that were registered using {@link TransientRegistry}
 */
public static void compactTransient(FluoConfiguration fluoConfig) throws Exception {
  Connector conn = getConnector(fluoConfig);

  try (FluoClient client = FluoFactory.newClient(fluoConfig)) {
    SimpleConfiguration appConfig = client.getAppConfiguration();

    TransientRegistry transientRegistry = new TransientRegistry(appConfig);
    List<RowRange> ranges = transientRegistry.getTransientRanges();

    for (RowRange r : ranges) {
      long t1 = System.currentTimeMillis();
      conn.tableOperations().compact(fluoConfig.getAccumuloTable(),
          new Text(r.getStart().toArray()), new Text(r.getEnd().toArray()), true, true);
      long t2 = System.currentTimeMillis();
      logger.info("Compacted {} in {}ms", r, (t2 - t1));
    }
  }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * Prints Fluo table accessible using provided client
 *
 * @param client Fluo client to table
 */
public static void printFluoTable(FluoClient client) {
  try (Snapshot s = client.newSnapshot()) {
    System.out.println("== fluo start ==");
    for (RowColumnValue rcv : s.scanner().build()) {
      StringBuilder sb = new StringBuilder();
      Hex.encNonAscii(sb, rcv.getRow());
      sb.append(" ");
      Hex.encNonAscii(sb, rcv.getColumn(), " ");
      sb.append("\t");
      Hex.encNonAscii(sb, rcv.getValue());

      System.out.println(sb.toString());
    }
    System.out.println("=== fluo end ===");
  }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * A utility method to get all registered table optimizations. Many recipes will automatically
 * register table optimizations when configured.
 */
public static TableOptimizations getConfiguredOptimizations(FluoConfiguration fluoConfig) {
  try (FluoClient client = FluoFactory.newClient(fluoConfig)) {
    SimpleConfiguration appConfig = client.getAppConfiguration();
    TableOptimizations tableOptim = new TableOptimizations();

    SimpleConfiguration subset = appConfig.subset(PREFIX.substring(0, PREFIX.length() - 1));
    Iterator<String> keys = subset.getKeys();
    while (keys.hasNext()) {
      String key = keys.next();
      String clazz = subset.getString(key);
      try {
        TableOptimizationsFactory factory =
            Class.forName(clazz).asSubclass(TableOptimizationsFactory.class).newInstance();
        tableOptim.merge(factory.getTableOptimizations(key, appConfig));
      } catch (InstantiationException | IllegalAccessException | ClassNotFoundException e) {
        throw new RuntimeException(e);
      }
    }

    return tableOptim;
  }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
private Map<String, Long> getComputedWordCounts(FluoClient fc) {
  Map<String, Long> counts = new HashMap<>();

  try (Snapshot snap = fc.newSnapshot()) {

    CellScanner scanner = snap.scanner().over(Span.prefix("iwc:")).build();

    for (RowColumnValue rcv : scanner) {
      String[] tokens = rcv.getsRow().split(":");
      String word = tokens[2];
      Long count = Long.valueOf(tokens[1]);

      Assert.assertFalse("Word seen twice in index " + word, counts.containsKey(word));

      counts.put(word, count);
    }
  }

  return counts;
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
private Map<String, Long> computeWordCounts(FluoClient fc) {
  Map<String, Long> counts = new HashMap<>();

  try (Snapshot snap = fc.newSnapshot()) {

    CellScanner scanner =
        snap.scanner().over(Span.prefix("d:")).fetch(new Column("content", "current")).build();

    for (RowColumnValue rcv : scanner) {
      String[] words = rcv.getsValue().split("\\s+");
      for (String word : words) {
        if (word.isEmpty()) {
          continue;
        }

        counts.merge(word, 1L, Long::sum);
      }
    }
  }

  return counts;
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
@Test
public void testStress() throws Exception {
  try (FluoClient fc = FluoFactory.newClient(miniFluo.getClientConfiguration())) {
    Random rand = new Random();

    try (LoaderExecutor loader = fc.newLoaderExecutor()) {
      for (int i = 0; i < 1000; i++) {
        loader.execute(new DocumentLoader(randDocId(rand), randomDocument(rand)));
      }
    }

    miniFluo.waitForObservers();
    assertWordCountsEqual(fc);

    try (LoaderExecutor loader = fc.newLoaderExecutor()) {
      for (int i = 0; i < 100; i++) {
        loader.execute(new DocumentLoader(randDocId(rand), randomDocument(rand)));
      }
    }

    miniFluo.waitForObservers();
    assertWordCountsEqual(fc);
  }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
protected Map<String, Set<String>> getFluoReferees(FluoClient fc) {
  Map<String, Set<String>> fluoReferees = new HashMap<>();

  try (Snapshot snap = fc.newSnapshot()) {

    Column currCol = new Column("content", "current");
    RowScanner rowScanner = snap.scanner().over(Span.prefix("d:")).fetch(currCol).byRow().build();

    for (ColumnScanner columnScanner : rowScanner) {
      String docid = columnScanner.getsRow().substring(2);

      for (ColumnValue columnValue : columnScanner) {
        String[] refs = columnValue.getsValue().split(" ");

        for (String ref : refs) {
          if (ref.isEmpty())
            continue;

          fluoReferees.computeIfAbsent(ref, k -> new HashSet<>()).add(docid);
        }
      }
    }
  }
  return fluoReferees;
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
private static FluoClient createFluoClient(final PcjAdminClientProperties clientProps) {
    checkNotNull(clientProps);
    final FluoConfiguration fluoConfig = new FluoConfiguration();

    // Fluo configuration values.
    fluoConfig.setApplicationName( clientProps.getFluoAppName() );
    fluoConfig.setInstanceZookeepers( clientProps.getAccumuloZookeepers() +  "/fluo" );

    // Accumulo Connection Stuff.
    fluoConfig.setAccumuloZookeepers( clientProps.getAccumuloZookeepers() );
    fluoConfig.setAccumuloInstance( clientProps.getAccumuloInstance() );
    fluoConfig.setAccumuloUser( clientProps.getAccumuloUsername() );
    fluoConfig.setAccumuloPassword( clientProps.getAccumuloPassword() );

    // Connect the client.
    return FluoFactory.newClient(fluoConfig);
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * Create a PeriodicNotificationApplication.
 * @param conf - Configuration object that specifies the parameters needed to create the application
 * @return PeriodicNotificationApplication to periodically poll Rya Fluo for new results
 * @throws PeriodicApplicationException
 */
public static PeriodicNotificationApplication getPeriodicApplication(final PeriodicNotificationApplicationConfiguration conf) throws PeriodicApplicationException {
    final Properties kafkaConsumerProps = getKafkaConsumerProperties(conf);
    final Properties kafkaProducerProps = getKafkaProducerProperties(conf);

    final BlockingQueue<TimestampedNotification> notifications = new LinkedBlockingQueue<>();
    final BlockingQueue<NodeBin> bins = new LinkedBlockingQueue<>();
    final BlockingQueue<BindingSetRecord> bindingSets = new LinkedBlockingQueue<>();

    FluoClient fluo = null;
    try {
        final PeriodicQueryResultStorage storage = getPeriodicQueryResultStorage(conf);
        fluo = FluoClientFactory.getFluoClient(conf.getFluoAppName(), Optional.of(conf.getFluoTableName()), conf);
        final NotificationCoordinatorExecutor coordinator = getCoordinator(conf.getCoordinatorThreads(), notifications);
        addRegisteredNotices(coordinator, fluo.newSnapshot());
        final KafkaExporterExecutor exporter = getExporter(conf.getExporterThreads(), kafkaProducerProps, bindingSets);
        final PeriodicQueryPrunerExecutor pruner = getPruner(storage, fluo, conf.getPrunerThreads(), bins);
        final NotificationProcessorExecutor processor = getProcessor(storage, notifications, bins, bindingSets, conf.getProcessorThreads());
        final KafkaNotificationProvider provider = getProvider(conf.getProducerThreads(), conf.getNotificationTopic(), coordinator, kafkaConsumerProps);
        return PeriodicNotificationApplication.builder().setCoordinator(coordinator).setProvider(provider).setExporter(exporter)
                .setProcessor(processor).setPruner(pruner).build();
    } catch (AccumuloException | AccumuloSecurityException e) {
        throw new PeriodicApplicationException(e.getMessage());
    }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * Creates a FluoClient
 * @param appName - name of Fluo application
 * @param tableName - name of Fluo table
 * @param conf - AccumuloConfiguration (must contain Accumulo User, Accumulo Instance, Accumulo Password, and Accumulo Zookeepers)
 * @return FluoClient for connecting to Fluo
 */
public static FluoClient getFluoClient(final String appName, final Optional<String> tableName, final AccumuloRdfConfiguration conf) {
    final FluoConfiguration fluoConfig = new FluoConfiguration();
    fluoConfig.setAccumuloInstance(conf.getAccumuloInstance());
    fluoConfig.setAccumuloUser(conf.getAccumuloUser());
    fluoConfig.setAccumuloPassword(conf.getAccumuloPassword());
    fluoConfig.setInstanceZookeepers(conf.getAccumuloZookeepers() + "/fluo");
    fluoConfig.setAccumuloZookeepers(conf.getAccumuloZookeepers());
    fluoConfig.setApplicationName(appName);
    if (tableName.isPresent()) {
        fluoConfig.setAccumuloTable(tableName.get());
    } else {
        fluoConfig.setAccumuloTable(appName);
    }
    return FluoFactory.newClient(fluoConfig);
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
@Override
public Optional<IncrementalResultExporter> build(Context context) throws IncrementalExporterFactoryException, ConfigurationException {
    Preconditions.checkNotNull(context);
    
    RyaSubGraphExportParameters params = new RyaSubGraphExportParameters(context.getObserverConfiguration().toMap());

    if (params.getUseRyaSubGraphExporter()) {
        try {
            //Get FluoConfiguration from params
            FluoConfiguration conf = params.getFluoConfiguration();
            FluoClient fluo = FluoFactory.newClient(conf);
            
            //Create exporter
            RyaSubGraphExporter exporter = new RyaSubGraphExporter(fluo);
            return Optional.of(exporter);
        } catch (Exception e) {
            throw new IncrementalExporterFactoryException("Could not initialize the RyaSubGraphExporter", e);
        }
    }
    return Optional.absent();
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
private void createSpanBatches(FluoClient fluoClient, List<String> ids, List<String> prefixes, int batchSize) {

        Preconditions.checkArgument(ids.size() == prefixes.size());

        try (Transaction tx = fluoClient.newTransaction()) {
            for (int i = 0; i < ids.size(); i++) {
                String id = ids.get(i);
                String bsPrefix = prefixes.get(i);
                URI uri = vf.createURI(bsPrefix);
                Bytes prefixBytes = BindingHashShardingFunction.getShardedScanPrefix(id, uri);
                NodeType type = NodeType.fromNodeId(id).get();
                Column bsCol = type.getResultColumn();
                SpanBatchDeleteInformation.Builder builder = SpanBatchDeleteInformation.builder().setBatchSize(batchSize)
                        .setColumn(bsCol);
                if (type == NodeType.JOIN) {
                    builder.setSpan(Span.prefix(type.getNodeTypePrefix()));
                    builder.setNodeId(java.util.Optional.of(id));
                } else {
                    builder.setSpan(Span.prefix(prefixBytes));
                }
                BatchInformationDAO.addBatch(tx, id, builder.build());
            }
            tx.commit();
        }
    }
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
private int countResults(FluoClient fluoClient, String nodeId, Column bsColumn) {
    try (Transaction tx = fluoClient.newTransaction()) {
        int count = 0;
        Optional<NodeType> type = NodeType.fromNodeId(nodeId);
        Bytes prefixBytes = Bytes.of(type.get().getNodeTypePrefix());
        RowScanner scanner = tx.scanner().over(Span.prefix(prefixBytes)).fetch(bsColumn).byRow().build();
        Iterator<ColumnScanner> colScanners = scanner.iterator();
        while (colScanners.hasNext()) {
            ColumnScanner colScanner = colScanners.next();
            BindingSetRow bsRow = BindingSetRow.makeFromShardedRow(prefixBytes, colScanner.getRow());
            if (bsRow.getNodeId().equals(nodeId)) {
                Iterator<ColumnValue> vals = colScanner.iterator();
                while (vals.hasNext()) {
                    vals.next();
                    count++;
                }
            }
        }
        tx.commit();
        return count;
    }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
private void verifyCounts(FluoClient fluoClient, List<String> ids, List<Integer> expectedCounts) {
    Preconditions.checkArgument(ids.size() == expectedCounts.size());
    for (int i = 0; i < ids.size(); i++) {
        String id = ids.get(i);
        int expected = expectedCounts.get(i);
        NodeType type = NodeType.fromNodeId(id).get();
        int count = countResults(fluoClient, id, type.getResultColumn());
        switch (type) {
        case STATEMENT_PATTERN:
            assertEquals(expected, count);
            break;
        case JOIN:
            assertEquals(expected, count);
            break;
        case QUERY:
            assertEquals(expected, count);
            break;
        default:
            break;
        }
    }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
@Test
public void testSparkThenFluoIndexing() throws Exception {

  Map<URL, Page> pageMap = readPages(new File("src/test/resources/wat-18.warc"));
  List<Page> pages = new ArrayList<>(pageMap.values());

  env.initializeIndexes(ctx, ctx.parallelize(pages.subList(0, 2)), new IndexStats(ctx));

  assertOutput(pages.subList(0, 2));

  try (FluoClient client = FluoFactory.newClient(getMiniFluo().getClientConfiguration());
      LoaderExecutor le = client.newLoaderExecutor()) {
    for (Page page : pages.subList(2, pages.size())) {
      log.debug("Loading page {} with {} links", page.getUrl(), page.getOutboundLinks().size());
      le.execute(PageLoader.updatePage(page));
    }
  }
  getMiniFluo().waitForObservers();

  assertOutput(pages);
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * This test ensures that when there are PCJ tables in Accumulo as well as
 * the Fluo table's export destinations column, the command for fetching the
 * list of queries only includes queries that appear in both places.
 */
@Test
public void getQueryIds() throws AccumuloException, AccumuloSecurityException, TableExistsException {
    try(FluoClient fluoClient = FluoFactory.newClient(super.getFluoConfiguration())) {
        // Store a few SPARQL/Query ID pairs in the Fluo table.
        try(Transaction tx = fluoClient.newTransaction()) {
            tx.set("SPARQL_3", QUERY_NODE_ID, "ID_3");
            tx.set("SPARQL_1", QUERY_NODE_ID, "ID_1");
            tx.set("SPARQL_4", QUERY_NODE_ID, "ID_4");
            tx.set("SPARQL_2", QUERY_NODE_ID, "ID_2");
            tx.commit();
        }

        // Ensure the correct list of Query IDs is retured.
        final List<String> expected = Lists.newArrayList("ID_1", "ID_2", "ID_3", "ID_4");
        final List<String> queryIds = new ListQueryIds().listQueryIds(fluoClient);
        assertEquals(expected, queryIds);
    }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
@Test
public void countStatements() {
    // Insert some Triples into the Fluo app.
    final List<RyaStatement> triples = new ArrayList<>();
    triples.add( RyaStatement.builder().setSubject(new RyaURI("http://Alice")).setPredicate(new RyaURI("http://talksTo")).setObject(new RyaURI("http://Bob")).build() );
    triples.add( RyaStatement.builder().setSubject(new RyaURI("http://Bob")).setPredicate(new RyaURI("http://talksTo")).setObject(new RyaURI("http://Alice")).build() );
    triples.add( RyaStatement.builder().setSubject(new RyaURI("http://Charlie")).setPredicate(new RyaURI("http://talksTo")).setObject(new RyaURI("http://Bob")).build() );
    triples.add( RyaStatement.builder().setSubject(new RyaURI("http://David")).setPredicate(new RyaURI("http://talksTo")).setObject(new RyaURI("http://Bob")).build() );
    triples.add( RyaStatement.builder().setSubject(new RyaURI("http://Eve")).setPredicate(new RyaURI("http://talksTo")).setObject(new RyaURI("http://Bob")).build() );

    try(FluoClient fluoClient = FluoFactory.newClient(super.getFluoConfiguration())) {
        new InsertTriples().insert(fluoClient, triples, Optional.<String>absent());

        // Load some statements into the Fluo app.
        final BigInteger count = new CountStatements().countStatements(fluoClient);

        // Ensure the count matches the expected values.
        assertEquals(BigInteger.valueOf(5), count);
    }
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * Tells the Fluo PCJ Updater application to maintain a new PCJ.  This method provides
 * no guarantees that a PCJ with the given pcjId exists outside of Fluo. This method merely
 * creates the FluoQuery (metadata) inside of Fluo so that results can be incrementally generated
 * inside of Fluo.  Results are exported according to the Set of {@link ExportStrategy} enums.  If
 * the Rya ExportStrategy is specified, care should be taken to verify that the PCJ table exists.
 *
 * @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
 * @param sparql - sparql query String to be registered with Fluo
 * @param strategies - ExportStrategies used to specify how final results will be handled
 * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
 * @return The metadata that was written to the Fluo application for the PCJ.
 * @throws UnsupportedQueryException 
 * @throws MalformedQueryException
 */
public FluoQuery createPcj(
        final String pcjId,
        final String sparql,
        final Set<ExportStrategy> strategies,
        final FluoClient fluo) throws MalformedQueryException, UnsupportedQueryException {
    requireNonNull(pcjId);
    requireNonNull(sparql);
    requireNonNull(strategies);
    requireNonNull(fluo);

    FluoQuery fluoQuery = makeFluoQuery(sparql, pcjId, strategies);
    writeFluoQuery(fluo, fluoQuery, pcjId);

    return fluoQuery;
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * Tells the Fluo PCJ Updater application to maintain a new PCJ.
 * <p>
 * This call scans Rya for Statement Pattern matches and inserts them into
 * the Fluo application. It is assumed that results for any query registered
 * using this method will be exported to Kafka according to the Kafka {@link ExportStrategy}.
 *
 * @param sparql - sparql query that will registered with Fluo. (not null)
 * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
 * @param accumulo - Accumulo connector for connecting with Accumulo
 * @param ryaInstance - Name of Rya instance to connect to
 * @return The Fluo application's Query ID of the query that was created.
 * @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
 * @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
 * @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
 * @throws UnsupportedQueryException 
 */
public String withRyaIntegration(
        final String sparql,
        final FluoClient fluo,
        final Connector accumulo,
        final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException, UnsupportedQueryException {
    requireNonNull(sparql);
    requireNonNull(fluo);
    requireNonNull(accumulo);
    requireNonNull(ryaInstance);

    
    // Write the SPARQL query's structure to the Fluo Application.
    final FluoQuery fluoQuery = createPcj(sparql, fluo);
    //import results already ingested into Rya that match query
    importHistoricResultsIntoFluo(fluo, fluoQuery, accumulo, ryaInstance);
    // return queryId to the caller for later monitoring from the export.
    return fluoQuery.getQueryMetadata().getNodeId();
}
 

import org.apache.fluo.api.client.FluoClient; //导入依赖的package包/类
/**
 * Deletes all metadata and {@link BindingSet}s associated with a Rya
 * Precomputed Join Index from the Fluo application that is incrementally
 * updating it.
 *
 * @param client - Connects to the Fluo application that is updating the PCJ
 *            Index. (not null)
 * @param pcjId - The PCJ ID for the query that will removed from the Fluo
 *            application. (not null)
 * @throws UnsupportedQueryException - thrown when Fluo app is unable to read FluoQuery associated
 * with given pcjId.
 */
public void deletePcj(final FluoClient client, final String pcjId) throws UnsupportedQueryException {
    requireNonNull(client);
    requireNonNull(pcjId);

    final Transaction tx = client.newTransaction();

    // Delete the query's metadata. This halts input.
    final List<String> nodeIds = getNodeIds(tx, pcjId);
    deleteMetadata(tx, nodeIds, pcjId);

    // Delete the binding sets associated with the query's nodes.
    for (final String nodeId : nodeIds) {
        deleteData(client, nodeId);
    }
}