Make the Loader able to create the chronicle db properly
- Id
- cd0f5a6f26651bde5741be819fe678dbf106c70b
- Author
- Caio
- Commit time
- 2019-01-31T13:27:39+01:00
Modified pom.xml
<argument>-Xms1024m</argument>
<argument>-Dlogback.configurationFile=src/test/resources/logback-test.xml</argument>
<argument>co.caio.cerberus.Loader</argument>
- <argument>src/main/resources/application.properties</argument>
+ <argument>src/main/resources/cerberus.properties</argument>
<argument>tmp/document.jsonlines</argument>
</arguments>
</configuration>
Modified src/main/resources/cerberus.properties
cerberus.timeout=2s
cerberus.page-size=20
-cerberus.lucene.directory=tmp/hue
+cerberus.lucene.directory=tmp/lucene
cerberus.lmdb.directory=tmp/lmdb
cerberus.lmdb.max-size-mb=3000
-cerberus.chronicle.filename=tmp/chronicledb
+cerberus.chronicle.filename=tmp/chronicle.db
Modified src/main/java/co/caio/cerberus/Loader.java
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.stream.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
public class Loader {
private static final Logger logger = LoggerFactory.getLogger(Loader.class);
private final Path recipesFile;
- private final Path dataDir;
- private final Path dbDir;
- private final int maxDbSize;
+ private final Path luceneIndexDir;
+ private final Path chronicleFilename;
private final Serializer serializer = new Serializer();
private final String movePrefix;
var conf = new Properties();
conf.load(Files.newBufferedReader(configFile));
- dataDir = Path.of(conf.getProperty("cerberus.search.location"));
- dbDir = Path.of(conf.getProperty("cerberus.search.lmdb-location"));
- maxDbSize = Integer.parseInt(conf.getProperty("cerberus.search.lmdb-max-size"));
+ luceneIndexDir = Path.of(conf.getProperty("cerberus.lucene.directory"));
+ chronicleFilename = Path.of(conf.getProperty("cerberus.chronicle.filename"));
- assert !dataDir.toFile().isFile() && !dbDir.toFile().isFile() && maxDbSize > 0;
+ assert !luceneIndexDir.toFile().isFile() && chronicleFilename.toFile().isFile();
movePrefix = "old." + System.currentTimeMillis() + ".";
}
return Files.lines(recipesFile).map(serializer::readRecipe).flatMap(Optional::stream);
}
- public void createIndex() throws IOException {
+ void createIndex() throws IOException {
- moveExistingDirs(dataDir);
- dataDir.toFile().mkdirs();
+ moveExistingDirs(luceneIndexDir);
+ luceneIndexDir.toFile().mkdirs();
logger.info("Initializing indexer");
var indexer =
new Indexer.Builder()
.analyzer(IndexConfiguration.getAnalyzer())
- .dataDirectory(dataDir)
+ .dataDirectory(luceneIndexDir)
.createMode()
.build();
logger.info("Finished creating lucene index");
}
- public void createDatabase() throws IOException {
+ void createDatabase() throws Exception {
- moveExistingDirs(dbDir);
- // dbDir.toFile().mkdirs();
+ moveExistingFile(chronicleFilename);
- // FIXME open for creation
- logger.info("Creating metadata database at {}", dbDir);
- var db = ChronicleRecipeMetadataDatabase.open(dbDir);
+ var settings = computeChronicleRWSettings();
+ logger.info("Creating metadata database at {}", chronicleFilename);
+ var db =
+ ChronicleRecipeMetadataDatabase.create(chronicleFilename, settings.avg, settings.count);
- recipeStream()
- .map(RecipeMetadata::fromRecipe)
- .parallel()
- .forEach(rm -> db.saveAll(List.of(rm)));
- // Flux.fromStream(recipeStream().map(RecipeMetadata::fromRecipe))
- // .buffer(100_000)
- // .subscribe(
- // recipes -> {
- // logger.info("Writing a batch of {} recipes", recipes.size());
- // db.saveAll(recipes);
- // });
+ Flux.fromStream(recipeStream().map(RecipeMetadata::fromRecipe))
+ .buffer(100_000)
+ .subscribe(
+ recipes -> {
+ logger.info("Writing a batch of {} recipes", recipes.size());
+ db.saveAll(recipes);
+ });
+
+ db.close();
logger.info("Finished creating metadata database");
+ }
+
+ private void moveExistingFile(Path filename) {
+ if (!filename.toFile().exists()) {
+ return;
+ }
+
+ var newName = movePrefix + filename.getFileName();
+ var dest = filename.resolveSibling(newName);
+
+ if (filename.toFile().renameTo(dest.toFile())) {
+ logger.info("Renamed {} to {}", filename, dest);
+ } else {
+ logger.error("Failed renaming {} to {}", filename, dest);
+ throw new RuntimeException("Unexpected error. Aborting...");
+ }
}
private void moveExistingDirs(Path... paths) {
}
}
- public void averageValueSize() throws Exception {
- var avg =
- recipeStream()
- .parallel()
- .map(FlatBufferSerializer.INSTANCE::flattenRecipe)
- .mapToInt(bb -> bb.limit() - bb.position())
- .average();
- System.out.println(avg);
+ ChronicleRWSettings computeChronicleRWSettings() throws Exception {
+ Map<String, Long> computed = new HashMap<>();
+
+ logger.info("Computing ChronicleMap creation parameters");
+ recipeStream()
+ .parallel()
+ .map(FlatBufferSerializer.INSTANCE::flattenRecipe)
+ .forEach(
+ bb -> {
+ var bufferSize = bb.limit() - bb.position();
+
+ computed.compute("bytes", (key, orig) -> (orig == null ? 0 : orig) + bufferSize);
+
+ computed.compute("count", (key, orig) -> (orig == null ? 0 : orig) + 1);
+ });
+
+ System.out.println(computed);
+
+ long count = computed.get("count");
+ long bytes = computed.get("bytes");
+ double avg = ((double) bytes) / count;
+
+ logger.info("Computed averageValueByteSize={} and numRecipes={}", avg, count);
+ return new ChronicleRWSettings(avg, count);
+ }
+
+ class ChronicleRWSettings {
+ double avg;
+ long count;
+
+ ChronicleRWSettings(double avg, long count) {
+ this.avg = avg;
+ this.count = count;
+ }
}
public static void main(String[] args) throws Exception {
var loader = new Loader(Path.of(args[0]), Path.of(args[1]));
- // loader.averageValueSize();
loader.createDatabase();
- // loader.createIndex();
+ loader.createIndex();
logger.info("Done!");
}