@Test public void testRun() throws IOException { AssembleDictionaryTask task = new AssembleDictionaryTask(); task.setCloudService(service); task.setBucket("swetodblp-term-2part--1"); task.setTargetBucket("swetodblp-local"); task.setSchemaBucket("swetodblp1"); task.setSchemaFile("opus_august2007_closure.nt"); task.setTermStatsFile("term_stats.json"); task.setEncodedSchemaFile("opus_august2007_closure_encoded.csv"); task.setEncodedTermStatsFile("term_stats_encoded.json"); task.run(); }
@Override public void run() throws IOException { log.info("Assembling dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB"); Stopwatch stopwatch = Stopwatch.createStarted(); List<StorageObject> objects = this.cloudService.listCloudStorageObjects(bucket); List<Item> items = new ArrayList<>(); for(StorageObject object: objects) { String filename = object.getName(); if(filename.endsWith(FilenameUtils.KRYO_SERIALIZED_EXT)) { items.add(new Item(filename, object.getSize().longValue())); } } log.info("Found " + items.size() + ", serialized files"); int processors = Runtime.getRuntime().availableProcessors(); BinPackingPartition function = new BinPackingPartition(items); function.setMaxBinItems((long) processors); List<Partition> partitions = function.partition(); TermDictionary dictionary = TermDictionary.populateRDFOWLData(new TermDictionaryConcurrent()); List<Callable<Void>> tasks = getSubTasks(partitions, dictionary); try { if(tasks.size() == 1) { tasks.get(0).call(); } else if(processors == 1) { for(Callable<Void> task: tasks) { task.call(); } } else { ExecutorService executor = Utils.createFixedThreadPool(processors); try { executor.invokeAll(tasks); } finally { executor.shutdown(); } } tasks = null; } catch(Exception e) { log.error("Failed to process multiple files", e); throw new IOException(e); } int dicSize = dictionary.size(); log.info("Successfully assembled dictionary with size: " + dicSize + ", max resourceId: " + dictionary.getLargestResourceId() + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isNotBlank(this.schemaFile) && StringUtils.isNotBlank(this.schemaBucket)) { this.encodeSchema(dictionary); } if(StringUtils.isNotBlank(this.termStatsFile) && StringUtils.isNotBlank(this.encodedTermStatsFile)) { this.encodeTermsStats(dictionary); } if(StringUtils.isBlank(this.dictionaryFile)) { this.dictionaryFile = this.cloudService.getInstanceId() + '_' + FilenameUtils.getSerializedGZipedDictionaryFilename(); } this.dictionaryFile = FilenameUtils.getLocalFilePath(this.dictionaryFile); dictionary = ((ConcurrentDictionary) dictionary).getNonConcurrentDictionary(); log.info("Successfully created non concurrent dictionary for serialization, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); dictionary.toFile(dictionaryFile, true); dictionary = null; log.info("Successfully serialized dictionary with size: " + dicSize + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isBlank(this.targetBucket)) { this.targetBucket = bucket; } this.cloudService.uploadFileToCloudStorage(dictionaryFile, this.targetBucket); log.info("Successfully assembled, serialized and uploaded dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); }
AssembleDictionaryTask extends CommonTask { @Override public void run() throws IOException { log.info("Assembling dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB"); Stopwatch stopwatch = Stopwatch.createStarted(); List<StorageObject> objects = this.cloudService.listCloudStorageObjects(bucket); List<Item> items = new ArrayList<>(); for(StorageObject object: objects) { String filename = object.getName(); if(filename.endsWith(FilenameUtils.KRYO_SERIALIZED_EXT)) { items.add(new Item(filename, object.getSize().longValue())); } } log.info("Found " + items.size() + ", serialized files"); int processors = Runtime.getRuntime().availableProcessors(); BinPackingPartition function = new BinPackingPartition(items); function.setMaxBinItems((long) processors); List<Partition> partitions = function.partition(); TermDictionary dictionary = TermDictionary.populateRDFOWLData(new TermDictionaryConcurrent()); List<Callable<Void>> tasks = getSubTasks(partitions, dictionary); try { if(tasks.size() == 1) { tasks.get(0).call(); } else if(processors == 1) { for(Callable<Void> task: tasks) { task.call(); } } else { ExecutorService executor = Utils.createFixedThreadPool(processors); try { executor.invokeAll(tasks); } finally { executor.shutdown(); } } tasks = null; } catch(Exception e) { log.error("Failed to process multiple files", e); throw new IOException(e); } int dicSize = dictionary.size(); log.info("Successfully assembled dictionary with size: " + dicSize + ", max resourceId: " + dictionary.getLargestResourceId() + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isNotBlank(this.schemaFile) && StringUtils.isNotBlank(this.schemaBucket)) { this.encodeSchema(dictionary); } if(StringUtils.isNotBlank(this.termStatsFile) && StringUtils.isNotBlank(this.encodedTermStatsFile)) { this.encodeTermsStats(dictionary); } if(StringUtils.isBlank(this.dictionaryFile)) { this.dictionaryFile = this.cloudService.getInstanceId() + '_' + FilenameUtils.getSerializedGZipedDictionaryFilename(); } this.dictionaryFile = FilenameUtils.getLocalFilePath(this.dictionaryFile); dictionary = ((ConcurrentDictionary) dictionary).getNonConcurrentDictionary(); log.info("Successfully created non concurrent dictionary for serialization, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); dictionary.toFile(dictionaryFile, true); dictionary = null; log.info("Successfully serialized dictionary with size: " + dicSize + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isBlank(this.targetBucket)) { this.targetBucket = bucket; } this.cloudService.uploadFileToCloudStorage(dictionaryFile, this.targetBucket); log.info("Successfully assembled, serialized and uploaded dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); } }
AssembleDictionaryTask extends CommonTask { @Override public void run() throws IOException { log.info("Assembling dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB"); Stopwatch stopwatch = Stopwatch.createStarted(); List<StorageObject> objects = this.cloudService.listCloudStorageObjects(bucket); List<Item> items = new ArrayList<>(); for(StorageObject object: objects) { String filename = object.getName(); if(filename.endsWith(FilenameUtils.KRYO_SERIALIZED_EXT)) { items.add(new Item(filename, object.getSize().longValue())); } } log.info("Found " + items.size() + ", serialized files"); int processors = Runtime.getRuntime().availableProcessors(); BinPackingPartition function = new BinPackingPartition(items); function.setMaxBinItems((long) processors); List<Partition> partitions = function.partition(); TermDictionary dictionary = TermDictionary.populateRDFOWLData(new TermDictionaryConcurrent()); List<Callable<Void>> tasks = getSubTasks(partitions, dictionary); try { if(tasks.size() == 1) { tasks.get(0).call(); } else if(processors == 1) { for(Callable<Void> task: tasks) { task.call(); } } else { ExecutorService executor = Utils.createFixedThreadPool(processors); try { executor.invokeAll(tasks); } finally { executor.shutdown(); } } tasks = null; } catch(Exception e) { log.error("Failed to process multiple files", e); throw new IOException(e); } int dicSize = dictionary.size(); log.info("Successfully assembled dictionary with size: " + dicSize + ", max resourceId: " + dictionary.getLargestResourceId() + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isNotBlank(this.schemaFile) && StringUtils.isNotBlank(this.schemaBucket)) { this.encodeSchema(dictionary); } if(StringUtils.isNotBlank(this.termStatsFile) && StringUtils.isNotBlank(this.encodedTermStatsFile)) { this.encodeTermsStats(dictionary); } if(StringUtils.isBlank(this.dictionaryFile)) { this.dictionaryFile = this.cloudService.getInstanceId() + '_' + FilenameUtils.getSerializedGZipedDictionaryFilename(); } this.dictionaryFile = FilenameUtils.getLocalFilePath(this.dictionaryFile); dictionary = ((ConcurrentDictionary) dictionary).getNonConcurrentDictionary(); log.info("Successfully created non concurrent dictionary for serialization, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); dictionary.toFile(dictionaryFile, true); dictionary = null; log.info("Successfully serialized dictionary with size: " + dicSize + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isBlank(this.targetBucket)) { this.targetBucket = bucket; } this.cloudService.uploadFileToCloudStorage(dictionaryFile, this.targetBucket); log.info("Successfully assembled, serialized and uploaded dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); }  }
AssembleDictionaryTask extends CommonTask { @Override public void run() throws IOException { log.info("Assembling dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB"); Stopwatch stopwatch = Stopwatch.createStarted(); List<StorageObject> objects = this.cloudService.listCloudStorageObjects(bucket); List<Item> items = new ArrayList<>(); for(StorageObject object: objects) { String filename = object.getName(); if(filename.endsWith(FilenameUtils.KRYO_SERIALIZED_EXT)) { items.add(new Item(filename, object.getSize().longValue())); } } log.info("Found " + items.size() + ", serialized files"); int processors = Runtime.getRuntime().availableProcessors(); BinPackingPartition function = new BinPackingPartition(items); function.setMaxBinItems((long) processors); List<Partition> partitions = function.partition(); TermDictionary dictionary = TermDictionary.populateRDFOWLData(new TermDictionaryConcurrent()); List<Callable<Void>> tasks = getSubTasks(partitions, dictionary); try { if(tasks.size() == 1) { tasks.get(0).call(); } else if(processors == 1) { for(Callable<Void> task: tasks) { task.call(); } } else { ExecutorService executor = Utils.createFixedThreadPool(processors); try { executor.invokeAll(tasks); } finally { executor.shutdown(); } } tasks = null; } catch(Exception e) { log.error("Failed to process multiple files", e); throw new IOException(e); } int dicSize = dictionary.size(); log.info("Successfully assembled dictionary with size: " + dicSize + ", max resourceId: " + dictionary.getLargestResourceId() + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isNotBlank(this.schemaFile) && StringUtils.isNotBlank(this.schemaBucket)) { this.encodeSchema(dictionary); } if(StringUtils.isNotBlank(this.termStatsFile) && StringUtils.isNotBlank(this.encodedTermStatsFile)) { this.encodeTermsStats(dictionary); } if(StringUtils.isBlank(this.dictionaryFile)) { this.dictionaryFile = this.cloudService.getInstanceId() + '_' + FilenameUtils.getSerializedGZipedDictionaryFilename(); } this.dictionaryFile = FilenameUtils.getLocalFilePath(this.dictionaryFile); dictionary = ((ConcurrentDictionary) dictionary).getNonConcurrentDictionary(); log.info("Successfully created non concurrent dictionary for serialization, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); dictionary.toFile(dictionaryFile, true); dictionary = null; log.info("Successfully serialized dictionary with size: " + dicSize + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isBlank(this.targetBucket)) { this.targetBucket = bucket; } this.cloudService.uploadFileToCloudStorage(dictionaryFile, this.targetBucket); log.info("Successfully assembled, serialized and uploaded dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); }  @Override void run(); List<Callable<Void>> getSubTasks(List<Partition> partitions, TermDictionary dictionary); String getBucket(); void setBucket(String bucket); String getTargetBucket(); void setTargetBucket(String targetBucket); String getSchemaBucket(); void setSchemaBucket(String schemaBucket); String getSchemaFile(); void setSchemaFile(String schemaFile); String getTermStatsFile(); void setTermStatsFile(String termStatsFile); String getEncodedSchemaFile(); void setEncodedSchemaFile(String encodedSchemaFile); String getEncodedTermStatsFile(); void setEncodedTermStatsFile(String encodedTermStatsFile); String getDictionaryFile(); void setDictionaryFile(String dictionaryFile); }
AssembleDictionaryTask extends CommonTask { @Override public void run() throws IOException { log.info("Assembling dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB"); Stopwatch stopwatch = Stopwatch.createStarted(); List<StorageObject> objects = this.cloudService.listCloudStorageObjects(bucket); List<Item> items = new ArrayList<>(); for(StorageObject object: objects) { String filename = object.getName(); if(filename.endsWith(FilenameUtils.KRYO_SERIALIZED_EXT)) { items.add(new Item(filename, object.getSize().longValue())); } } log.info("Found " + items.size() + ", serialized files"); int processors = Runtime.getRuntime().availableProcessors(); BinPackingPartition function = new BinPackingPartition(items); function.setMaxBinItems((long) processors); List<Partition> partitions = function.partition(); TermDictionary dictionary = TermDictionary.populateRDFOWLData(new TermDictionaryConcurrent()); List<Callable<Void>> tasks = getSubTasks(partitions, dictionary); try { if(tasks.size() == 1) { tasks.get(0).call(); } else if(processors == 1) { for(Callable<Void> task: tasks) { task.call(); } } else { ExecutorService executor = Utils.createFixedThreadPool(processors); try { executor.invokeAll(tasks); } finally { executor.shutdown(); } } tasks = null; } catch(Exception e) { log.error("Failed to process multiple files", e); throw new IOException(e); } int dicSize = dictionary.size(); log.info("Successfully assembled dictionary with size: " + dicSize + ", max resourceId: " + dictionary.getLargestResourceId() + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isNotBlank(this.schemaFile) && StringUtils.isNotBlank(this.schemaBucket)) { this.encodeSchema(dictionary); } if(StringUtils.isNotBlank(this.termStatsFile) && StringUtils.isNotBlank(this.encodedTermStatsFile)) { this.encodeTermsStats(dictionary); } if(StringUtils.isBlank(this.dictionaryFile)) { this.dictionaryFile = this.cloudService.getInstanceId() + '_' + FilenameUtils.getSerializedGZipedDictionaryFilename(); } this.dictionaryFile = FilenameUtils.getLocalFilePath(this.dictionaryFile); dictionary = ((ConcurrentDictionary) dictionary).getNonConcurrentDictionary(); log.info("Successfully created non concurrent dictionary for serialization, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); dictionary.toFile(dictionaryFile, true); dictionary = null; log.info("Successfully serialized dictionary with size: " + dicSize + ", memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); if(StringUtils.isBlank(this.targetBucket)) { this.targetBucket = bucket; } this.cloudService.uploadFileToCloudStorage(dictionaryFile, this.targetBucket); log.info("Successfully assembled, serialized and uploaded dictionary, memory usage: " + Utils.getMemoryUsageInGB() + "GB" + ", timer: " + stopwatch); }  @Override void run(); List<Callable<Void>> getSubTasks(List<Partition> partitions, TermDictionary dictionary); String getBucket(); void setBucket(String bucket); String getTargetBucket(); void setTargetBucket(String targetBucket); String getSchemaBucket(); void setSchemaBucket(String schemaBucket); String getSchemaFile(); void setSchemaFile(String schemaFile); String getTermStatsFile(); void setTermStatsFile(String termStatsFile); String getEncodedSchemaFile(); void setEncodedSchemaFile(String encodedSchemaFile); String getEncodedTermStatsFile(); void setEncodedTermStatsFile(String encodedTermStatsFile); String getDictionaryFile(); void setDictionaryFile(String dictionaryFile);  }