ACCUMULO-2408 organize static final members of abstract class to a concrete implementation
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/4e725160 Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/4e725160 Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/4e725160 Branch: refs/heads/1.6.0-SNAPSHOT Commit: 4e7251601f9b51318d7e5373468dd852f12d0e0c Parents: b5b38d8 0bd62e3 Author: Eric Newton <eric.new...@gmail.com> Authored: Wed Feb 26 14:39:44 2014 -0500 Committer: Eric Newton <eric.new...@gmail.com> Committed: Wed Feb 26 14:39:44 2014 -0500 ---------------------------------------------------------------------- .../impl/TabletServerBatchReaderIterator.java | 12 +++---- .../client/impl/TabletServerBatchWriter.java | 8 ++--- .../core/client/impl/ThriftScanner.java | 4 +-- .../accumulo/core/client/impl/Translator.java | 9 ----- .../accumulo/core/client/impl/Translators.java | 37 ++++++++++++++++++++ .../accumulo/server/client/BulkImporter.java | 5 +-- .../apache/accumulo/tserver/TabletServer.java | 17 ++++----- .../mastermessage/SplitReportMessage.java | 3 +- 8 files changed, 63 insertions(+), 32 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/accumulo/blob/4e725160/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java ---------------------------------------------------------------------- diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java index 6ef514d,990a577..30e12c1 --- a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java +++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchReaderIterator.java @@@ -653,11 -647,10 +653,11 @@@ public class TabletServerBatchReaderIte TabletType ttype = TabletType.type(requested.keySet()); boolean waitForWrites = !ThriftScanner.serversWaitedForWrites.get(ttype).contains(server); - Map<TKeyExtent,List<TRange>> thriftTabletRanges = Translator.translate(requested, Translator.KET, new Translator.ListTranslator<Range,TRange>( - Translator.RT)); + Map<TKeyExtent,List<TRange>> thriftTabletRanges = Translator.translate(requested, Translators.KET, new Translator.ListTranslator<Range,TRange>( + Translators.RT)); - InitialMultiScan imsr = client.startMultiScan(Tracer.traceInfo(), credentials, thriftTabletRanges, Translator.translate(columns, Translators.CT), - options.serverSideIteratorList, options.serverSideIteratorOptions, ByteBufferUtil.toByteBuffers(authorizations.getAuthorizations()), waitForWrites); + InitialMultiScan imsr = client.startMultiScan(Tracer.traceInfo(), credentials.toThrift(instance), thriftTabletRanges, - Translator.translate(columns, Translator.CT), options.serverSideIteratorList, options.serverSideIteratorOptions, ++ Translator.translate(columns, Translators.CT), options.serverSideIteratorList, options.serverSideIteratorOptions, + ByteBufferUtil.toByteBuffers(authorizations.getAuthorizations()), waitForWrites); if (waitForWrites) ThriftScanner.serversWaitedForWrites.get(ttype).add(server); http://git-wip-us.apache.org/repos/asf/accumulo/blob/4e725160/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java ---------------------------------------------------------------------- diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java index 5586d9c,5b24f4a..76db4be --- a/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java +++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TabletServerBatchWriter.java @@@ -867,12 -867,12 +867,12 @@@ public class TabletServerBatchWriter Entry<KeyExtent,List<Mutation>> entry = tabMuts.entrySet().iterator().next(); try { - client.update(tinfo, credentials, entry.getKey().toThrift(), entry.getValue().get(0).toThrift()); + client.update(tinfo, credentials.toThrift(instance), entry.getKey().toThrift(), entry.getValue().get(0).toThrift()); } catch (NotServingTabletException e) { allFailures.addAll(entry.getKey().getTableId().toString(), entry.getValue()); - TabletLocator.getInstance(instance, new Text(entry.getKey().getTableId())).invalidateCache(entry.getKey()); + TabletLocator.getLocator(instance, new Text(entry.getKey().getTableId())).invalidateCache(entry.getKey()); } catch (ConstraintViolationException e) { - updatedConstraintViolations(Translator.translate(e.violationSummaries, Translator.TCVST)); + updatedConstraintViolations(Translator.translate(e.violationSummaries, Translators.TCVST)); } timeoutTracker.madeProgress(); } else { http://git-wip-us.apache.org/repos/asf/accumulo/blob/4e725160/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java ---------------------------------------------------------------------- diff --cc core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java index efb31e8,ccee661..b765ef4 --- a/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java +++ b/core/src/main/java/org/apache/accumulo/core/client/impl/ThriftScanner.java @@@ -96,9 -96,9 +96,9 @@@ public class ThriftScanner TabletType ttype = TabletType.type(extent); boolean waitForWrites = !serversWaitedForWrites.get(ttype).contains(server); - InitialScan isr = client.startScan(tinfo, scanState.credentials, extent.toThrift(), scanState.range.toThrift(), + InitialScan isr = client.startScan(tinfo, scanState.credentials.toThrift(instance), extent.toThrift(), scanState.range.toThrift(), - Translator.translate(scanState.columns, Translator.CT), scanState.size, scanState.serverSideIteratorList, scanState.serverSideIteratorOptions, + Translator.translate(scanState.columns, Translators.CT), scanState.size, scanState.serverSideIteratorList, scanState.serverSideIteratorOptions, - scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated); + scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated, scanState.readaheadThreshold); if (waitForWrites) serversWaitedForWrites.get(ttype).add(server); @@@ -396,9 -384,9 +396,9 @@@ TabletType ttype = TabletType.type(loc.tablet_extent); boolean waitForWrites = !serversWaitedForWrites.get(ttype).contains(loc.tablet_location); - InitialScan is = client.startScan(tinfo, scanState.credentials, loc.tablet_extent.toThrift(), scanState.range.toThrift(), + InitialScan is = client.startScan(tinfo, scanState.credentials.toThrift(scanState.instance), loc.tablet_extent.toThrift(), scanState.range.toThrift(), - Translator.translate(scanState.columns, Translator.CT), scanState.size, scanState.serverSideIteratorList, scanState.serverSideIteratorOptions, + Translator.translate(scanState.columns, Translators.CT), scanState.size, scanState.serverSideIteratorList, scanState.serverSideIteratorOptions, - scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated); + scanState.authorizations.getAuthorizationsBB(), waitForWrites, scanState.isolated, scanState.readaheadThreshold); if (waitForWrites) serversWaitedForWrites.get(ttype).add(loc.tablet_location); http://git-wip-us.apache.org/repos/asf/accumulo/blob/4e725160/core/src/main/java/org/apache/accumulo/core/client/impl/Translator.java ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/accumulo/blob/4e725160/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java ---------------------------------------------------------------------- diff --cc server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java index 8f35b23,0000000..7f59eae mode 100644,000000..100644 --- a/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java +++ b/server/base/src/main/java/org/apache/accumulo/server/client/BulkImporter.java @@@ -1,778 -1,0 +1,779 @@@ +/* + * 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. + */ +package org.apache.accumulo.server.client; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import org.apache.accumulo.core.client.AccumuloException; +import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.accumulo.core.client.Instance; +import org.apache.accumulo.core.client.impl.ServerClient; +import org.apache.accumulo.core.client.impl.TabletLocator; +import org.apache.accumulo.core.client.impl.TabletLocator.TabletLocation; +import org.apache.accumulo.core.client.impl.Translator; ++import org.apache.accumulo.core.client.impl.Translators; +import org.apache.accumulo.core.client.impl.thrift.ClientService; +import org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException; +import org.apache.accumulo.core.client.impl.thrift.ThriftTableOperationException; +import org.apache.accumulo.core.conf.AccumuloConfiguration; +import org.apache.accumulo.core.conf.Property; +import org.apache.accumulo.core.data.ByteSequence; +import org.apache.accumulo.core.data.KeyExtent; +import org.apache.accumulo.core.data.Range; +import org.apache.accumulo.core.data.thrift.TKeyExtent; +import org.apache.accumulo.core.file.FileOperations; +import org.apache.accumulo.core.file.FileSKVIterator; +import org.apache.accumulo.core.file.FileUtil; +import org.apache.accumulo.core.metadata.MetadataTable; +import org.apache.accumulo.core.security.Credentials; +import org.apache.accumulo.core.tabletserver.thrift.TabletClientService; +import org.apache.accumulo.core.util.CachedConfiguration; +import org.apache.accumulo.core.util.LoggingRunnable; +import org.apache.accumulo.core.util.NamingThreadFactory; +import org.apache.accumulo.core.util.StopWatch; +import org.apache.accumulo.core.util.ThriftUtil; +import org.apache.accumulo.core.util.UtilWaitThread; +import org.apache.accumulo.server.conf.ServerConfiguration; +import org.apache.accumulo.server.fs.VolumeManager; +import org.apache.accumulo.server.fs.VolumeManagerImpl; +import org.apache.accumulo.trace.instrument.TraceRunnable; +import org.apache.accumulo.trace.instrument.Tracer; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.io.Text; +import org.apache.log4j.Logger; +import org.apache.thrift.TServiceClient; + +public class BulkImporter { + + private static final Logger log = Logger.getLogger(BulkImporter.class); + + public static List<String> bulkLoad(AccumuloConfiguration conf, Instance instance, Credentials creds, long tid, String tableId, List<String> files, + String errorDir, boolean setTime) throws IOException, AccumuloException, AccumuloSecurityException, ThriftTableOperationException { + AssignmentStats stats = new BulkImporter(conf, instance, creds, tid, tableId, setTime).importFiles(files, new Path(errorDir)); + List<String> result = new ArrayList<String>(); + for (Path p : stats.completeFailures.keySet()) { + result.add(p.toString()); + } + return result; + } + + private StopWatch<Timers> timer; + + private static enum Timers { + EXAMINE_MAP_FILES, QUERY_METADATA, IMPORT_MAP_FILES, SLEEP, TOTAL + } + + private Instance instance; + private Credentials credentials; + private String tableId; + private long tid; + private AccumuloConfiguration acuConf; + private boolean setTime; + + public BulkImporter(AccumuloConfiguration conf, Instance instance, Credentials credentials, long tid, String tableId, boolean setTime) { + this.instance = instance; + this.credentials = credentials; + this.tid = tid; + this.tableId = tableId; + this.acuConf = conf; + this.setTime = setTime; + } + + public AssignmentStats importFiles(List<String> files, Path failureDir) throws IOException, AccumuloException, AccumuloSecurityException, + ThriftTableOperationException { + + int numThreads = acuConf.getCount(Property.TSERV_BULK_PROCESS_THREADS); + int numAssignThreads = acuConf.getCount(Property.TSERV_BULK_ASSIGNMENT_THREADS); + + timer = new StopWatch<Timers>(Timers.class); + timer.start(Timers.TOTAL); + + Configuration conf = CachedConfiguration.getInstance(); + VolumeManagerImpl.get(acuConf); + final VolumeManager fs = VolumeManagerImpl.get(acuConf); + + Set<Path> paths = new HashSet<Path>(); + for (String file : files) { + paths.add(new Path(file)); + } + AssignmentStats assignmentStats = new AssignmentStats(paths.size()); + + final Map<Path,List<KeyExtent>> completeFailures = Collections.synchronizedSortedMap(new TreeMap<Path,List<KeyExtent>>()); + + ClientService.Client client = null; + final TabletLocator locator = TabletLocator.getLocator(instance, new Text(tableId)); + + try { + final Map<Path,List<TabletLocation>> assignments = Collections.synchronizedSortedMap(new TreeMap<Path,List<TabletLocation>>()); + + timer.start(Timers.EXAMINE_MAP_FILES); + ExecutorService threadPool = Executors.newFixedThreadPool(numThreads, new NamingThreadFactory("findOverlapping")); + + for (Path path : paths) { + final Path mapFile = path; + Runnable getAssignments = new Runnable() { + @Override + public void run() { + List<TabletLocation> tabletsToAssignMapFileTo = Collections.emptyList(); + try { + tabletsToAssignMapFileTo = findOverlappingTablets(ServerConfiguration.getSystemConfiguration(instance), fs, locator, mapFile, credentials); + } catch (Exception ex) { + log.warn("Unable to find tablets that overlap file " + mapFile.toString()); + } + log.debug("Map file " + mapFile + " found to overlap " + tabletsToAssignMapFileTo.size() + " tablets"); + if (tabletsToAssignMapFileTo.size() == 0) { + List<KeyExtent> empty = Collections.emptyList(); + completeFailures.put(mapFile, empty); + } else + assignments.put(mapFile, tabletsToAssignMapFileTo); + } + }; + threadPool.submit(new TraceRunnable(new LoggingRunnable(log, getAssignments))); + } + threadPool.shutdown(); + while (!threadPool.isTerminated()) { + try { + threadPool.awaitTermination(60, TimeUnit.SECONDS); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + timer.stop(Timers.EXAMINE_MAP_FILES); + + assignmentStats.attemptingAssignments(assignments); + Map<Path,List<KeyExtent>> assignmentFailures = assignMapFiles(acuConf, instance, conf, credentials, fs, tableId, assignments, paths, numAssignThreads, + numThreads); + assignmentStats.assignmentsFailed(assignmentFailures); + + Map<Path,Integer> failureCount = new TreeMap<Path,Integer>(); + + for (Entry<Path,List<KeyExtent>> entry : assignmentFailures.entrySet()) + failureCount.put(entry.getKey(), 1); + + long sleepTime = 2 * 1000; + while (assignmentFailures.size() > 0) { + sleepTime = Math.min(sleepTime * 2, 60 * 1000); + locator.invalidateCache(); + // assumption about assignment failures is that it caused by a split + // happening or a missing location + // + // for splits we need to find children key extents that cover the + // same key range and are contiguous (no holes, no overlap) + + timer.start(Timers.SLEEP); + UtilWaitThread.sleep(sleepTime); + timer.stop(Timers.SLEEP); + + log.debug("Trying to assign " + assignmentFailures.size() + " map files that previously failed on some key extents"); + assignments.clear(); + + // for failed key extents, try to find children key extents to + // assign to + for (Entry<Path,List<KeyExtent>> entry : assignmentFailures.entrySet()) { + Iterator<KeyExtent> keListIter = entry.getValue().iterator(); + + List<TabletLocation> tabletsToAssignMapFileTo = new ArrayList<TabletLocation>(); + + while (keListIter.hasNext()) { + KeyExtent ke = keListIter.next(); + + try { + timer.start(Timers.QUERY_METADATA); + tabletsToAssignMapFileTo.addAll(findOverlappingTablets(ServerConfiguration.getSystemConfiguration(instance), fs, locator, entry.getKey(), ke, + credentials)); + timer.stop(Timers.QUERY_METADATA); + keListIter.remove(); + } catch (Exception ex) { + log.warn("Exception finding overlapping tablets, will retry tablet " + ke); + } + } + + if (tabletsToAssignMapFileTo.size() > 0) + assignments.put(entry.getKey(), tabletsToAssignMapFileTo); + } + + assignmentStats.attemptingAssignments(assignments); + Map<Path,List<KeyExtent>> assignmentFailures2 = assignMapFiles(acuConf, instance, conf, credentials, fs, tableId, assignments, paths, numAssignThreads, + numThreads); + assignmentStats.assignmentsFailed(assignmentFailures2); + + // merge assignmentFailures2 into assignmentFailures + for (Entry<Path,List<KeyExtent>> entry : assignmentFailures2.entrySet()) { + assignmentFailures.get(entry.getKey()).addAll(entry.getValue()); + + Integer fc = failureCount.get(entry.getKey()); + if (fc == null) + fc = 0; + + failureCount.put(entry.getKey(), fc + 1); + } + + // remove map files that have no more key extents to assign + Iterator<Entry<Path,List<KeyExtent>>> afIter = assignmentFailures.entrySet().iterator(); + while (afIter.hasNext()) { + Entry<Path,List<KeyExtent>> entry = afIter.next(); + if (entry.getValue().size() == 0) + afIter.remove(); + } + + Set<Entry<Path,Integer>> failureIter = failureCount.entrySet(); + for (Entry<Path,Integer> entry : failureIter) { + int retries = acuConf.getCount(Property.TSERV_BULK_RETRY); + if (entry.getValue() > retries && assignmentFailures.get(entry.getKey()) != null) { + log.error("Map file " + entry.getKey() + " failed more than " + retries + " times, giving up."); + completeFailures.put(entry.getKey(), assignmentFailures.get(entry.getKey())); + assignmentFailures.remove(entry.getKey()); + } + } + } + assignmentStats.assignmentsAbandoned(completeFailures); + Set<Path> failedFailures = processFailures(completeFailures); + assignmentStats.unrecoveredMapFiles(failedFailures); + + timer.stop(Timers.TOTAL); + printReport(); + return assignmentStats; + } finally { + if (client != null) + ServerClient.close(client); + locator.invalidateCache(); + } + } + + private void printReport() { + long totalTime = 0; + for (Timers t : Timers.values()) { + if (t == Timers.TOTAL) + continue; + + totalTime += timer.get(t); + } + + log.debug("BULK IMPORT TIMING STATISTICS"); + log.debug(String.format("Examine map files : %,10.2f secs %6.2f%s", timer.getSecs(Timers.EXAMINE_MAP_FILES), 100.0 * timer.get(Timers.EXAMINE_MAP_FILES) + / timer.get(Timers.TOTAL), "%")); + log.debug(String.format("Query %-14s : %,10.2f secs %6.2f%s", MetadataTable.NAME, timer.getSecs(Timers.QUERY_METADATA), + 100.0 * timer.get(Timers.QUERY_METADATA) / timer.get(Timers.TOTAL), "%")); + log.debug(String.format("Import Map Files : %,10.2f secs %6.2f%s", timer.getSecs(Timers.IMPORT_MAP_FILES), 100.0 * timer.get(Timers.IMPORT_MAP_FILES) + / timer.get(Timers.TOTAL), "%")); + log.debug(String.format("Sleep : %,10.2f secs %6.2f%s", timer.getSecs(Timers.SLEEP), + 100.0 * timer.get(Timers.SLEEP) / timer.get(Timers.TOTAL), "%")); + log.debug(String.format("Misc : %,10.2f secs %6.2f%s", (timer.get(Timers.TOTAL) - totalTime) / 1000.0, 100.0 + * (timer.get(Timers.TOTAL) - totalTime) / timer.get(Timers.TOTAL), "%")); + log.debug(String.format("Total : %,10.2f secs", timer.getSecs(Timers.TOTAL))); + } + + private Set<Path> processFailures(Map<Path,List<KeyExtent>> completeFailures) { + // we should check if map file was not assigned to any tablets, then we + // should just move it; not currently being done? + + Set<Entry<Path,List<KeyExtent>>> es = completeFailures.entrySet(); + + if (completeFailures.size() == 0) + return Collections.emptySet(); + + log.debug("The following map files failed "); + + for (Entry<Path,List<KeyExtent>> entry : es) { + List<KeyExtent> extents = entry.getValue(); + + for (KeyExtent keyExtent : extents) + log.debug("\t" + entry.getKey() + " -> " + keyExtent); + } + + return Collections.emptySet(); + } + + private class AssignmentInfo { + public AssignmentInfo(KeyExtent keyExtent, Long estSize) { + this.ke = keyExtent; + this.estSize = estSize; + } + + KeyExtent ke; + long estSize; + } + + private static List<KeyExtent> extentsOf(List<TabletLocation> locations) { + List<KeyExtent> result = new ArrayList<KeyExtent>(locations.size()); + for (TabletLocation tl : locations) + result.add(tl.tablet_extent); + return result; + } + + private Map<Path,List<AssignmentInfo>> estimateSizes(final AccumuloConfiguration acuConf, final Configuration conf, final VolumeManager vm, + Map<Path,List<TabletLocation>> assignments, Collection<Path> paths, int numThreads) { + + long t1 = System.currentTimeMillis(); + final Map<Path,Long> mapFileSizes = new TreeMap<Path,Long>(); + + try { + for (Path path : paths) { + FileSystem fs = vm.getFileSystemByPath(path); + mapFileSizes.put(path, fs.getContentSummary(path).getLength()); + } + } catch (IOException e) { + log.error("Failed to get map files in for " + paths + ": " + e.getMessage(), e); + throw new RuntimeException(e); + } + + final Map<Path,List<AssignmentInfo>> ais = Collections.synchronizedMap(new TreeMap<Path,List<AssignmentInfo>>()); + + ExecutorService threadPool = Executors.newFixedThreadPool(numThreads, new NamingThreadFactory("estimateSizes")); + + for (final Entry<Path,List<TabletLocation>> entry : assignments.entrySet()) { + if (entry.getValue().size() == 1) { + TabletLocation tabletLocation = entry.getValue().get(0); + + // if the tablet completely contains the map file, there is no + // need to estimate its + // size + ais.put(entry.getKey(), Collections.singletonList(new AssignmentInfo(tabletLocation.tablet_extent, mapFileSizes.get(entry.getKey())))); + continue; + } + + Runnable estimationTask = new Runnable() { + @Override + public void run() { + Map<KeyExtent,Long> estimatedSizes = null; + + try { + FileSystem fs = vm.getFileSystemByPath(entry.getKey()); + estimatedSizes = FileUtil.estimateSizes(acuConf, entry.getKey(), mapFileSizes.get(entry.getKey()), extentsOf(entry.getValue()), conf, fs); + } catch (IOException e) { + log.warn("Failed to estimate map file sizes " + e.getMessage()); + } + + if (estimatedSizes == null) { + // estimation failed, do a simple estimation + estimatedSizes = new TreeMap<KeyExtent,Long>(); + long estSize = (long) (mapFileSizes.get(entry.getKey()) / (double) entry.getValue().size()); + for (TabletLocation tl : entry.getValue()) + estimatedSizes.put(tl.tablet_extent, estSize); + } + + List<AssignmentInfo> assignmentInfoList = new ArrayList<AssignmentInfo>(estimatedSizes.size()); + + for (Entry<KeyExtent,Long> entry2 : estimatedSizes.entrySet()) + assignmentInfoList.add(new AssignmentInfo(entry2.getKey(), entry2.getValue())); + + ais.put(entry.getKey(), assignmentInfoList); + } + }; + + threadPool.submit(new TraceRunnable(new LoggingRunnable(log, estimationTask))); + } + + threadPool.shutdown(); + + while (!threadPool.isTerminated()) { + try { + threadPool.awaitTermination(60, TimeUnit.SECONDS); + } catch (InterruptedException e) { + e.printStackTrace(); + throw new RuntimeException(e); + } + } + + long t2 = System.currentTimeMillis(); + + log.debug(String.format("Estimated map files sizes in %6.2f secs", (t2 - t1) / 1000.0)); + + return ais; + } + + private static Map<KeyExtent,String> locationsOf(Map<Path,List<TabletLocation>> assignments) { + Map<KeyExtent,String> result = new HashMap<KeyExtent,String>(); + for (List<TabletLocation> entry : assignments.values()) { + for (TabletLocation tl : entry) { + result.put(tl.tablet_extent, tl.tablet_location); + } + } + return result; + } + + private Map<Path,List<KeyExtent>> assignMapFiles(AccumuloConfiguration acuConf, Instance instance, Configuration conf, Credentials credentials, + VolumeManager fs, String tableId, Map<Path,List<TabletLocation>> assignments, Collection<Path> paths, int numThreads, int numMapThreads) { + timer.start(Timers.EXAMINE_MAP_FILES); + Map<Path,List<AssignmentInfo>> assignInfo = estimateSizes(acuConf, conf, fs, assignments, paths, numMapThreads); + timer.stop(Timers.EXAMINE_MAP_FILES); + + Map<Path,List<KeyExtent>> ret; + + timer.start(Timers.IMPORT_MAP_FILES); + ret = assignMapFiles(credentials, tableId, assignInfo, locationsOf(assignments), numThreads); + timer.stop(Timers.IMPORT_MAP_FILES); + + return ret; + } + + private class AssignmentTask implements Runnable { + final Map<Path,List<KeyExtent>> assignmentFailures; + String location; + Credentials credentials; + private Map<KeyExtent,List<PathSize>> assignmentsPerTablet; + + public AssignmentTask(Credentials credentials, Map<Path,List<KeyExtent>> assignmentFailures, String tableName, String location, + Map<KeyExtent,List<PathSize>> assignmentsPerTablet) { + this.assignmentFailures = assignmentFailures; + this.location = location; + this.assignmentsPerTablet = assignmentsPerTablet; + this.credentials = credentials; + } + + private void handleFailures(Collection<KeyExtent> failures, String message) { + for (KeyExtent ke : failures) { + List<PathSize> mapFiles = assignmentsPerTablet.get(ke); + synchronized (assignmentFailures) { + for (PathSize pathSize : mapFiles) { + List<KeyExtent> existingFailures = assignmentFailures.get(pathSize.path); + if (existingFailures == null) { + existingFailures = new ArrayList<KeyExtent>(); + assignmentFailures.put(pathSize.path, existingFailures); + } + + existingFailures.add(ke); + } + } + + log.info("Could not assign " + mapFiles.size() + " map files to tablet " + ke + " because : " + message + ". Will retry ..."); + } + } + + @Override + public void run() { + HashSet<Path> uniqMapFiles = new HashSet<Path>(); + for (List<PathSize> mapFiles : assignmentsPerTablet.values()) + for (PathSize ps : mapFiles) + uniqMapFiles.add(ps.path); + + log.debug("Assigning " + uniqMapFiles.size() + " map files to " + assignmentsPerTablet.size() + " tablets at " + location); + + try { + List<KeyExtent> failures = assignMapFiles(credentials, location, assignmentsPerTablet); + handleFailures(failures, "Not Serving Tablet"); + } catch (AccumuloException e) { + handleFailures(assignmentsPerTablet.keySet(), e.getMessage()); + } catch (AccumuloSecurityException e) { + handleFailures(assignmentsPerTablet.keySet(), e.getMessage()); + } + } + + } + + private class PathSize { + public PathSize(Path mapFile, long estSize) { + this.path = mapFile; + this.estSize = estSize; + } + + Path path; + long estSize; + + @Override + public String toString() { + return path + " " + estSize; + } + } + + private Map<Path,List<KeyExtent>> assignMapFiles(Credentials credentials, String tableName, Map<Path,List<AssignmentInfo>> assignments, + Map<KeyExtent,String> locations, int numThreads) { + + // group assignments by tablet + Map<KeyExtent,List<PathSize>> assignmentsPerTablet = new TreeMap<KeyExtent,List<PathSize>>(); + for (Entry<Path,List<AssignmentInfo>> entry : assignments.entrySet()) { + Path mapFile = entry.getKey(); + List<AssignmentInfo> tabletsToAssignMapFileTo = entry.getValue(); + + for (AssignmentInfo ai : tabletsToAssignMapFileTo) { + List<PathSize> mapFiles = assignmentsPerTablet.get(ai.ke); + if (mapFiles == null) { + mapFiles = new ArrayList<PathSize>(); + assignmentsPerTablet.put(ai.ke, mapFiles); + } + + mapFiles.add(new PathSize(mapFile, ai.estSize)); + } + } + + // group assignments by tabletserver + + Map<Path,List<KeyExtent>> assignmentFailures = Collections.synchronizedMap(new TreeMap<Path,List<KeyExtent>>()); + + TreeMap<String,Map<KeyExtent,List<PathSize>>> assignmentsPerTabletServer = new TreeMap<String,Map<KeyExtent,List<PathSize>>>(); + + for (Entry<KeyExtent,List<PathSize>> entry : assignmentsPerTablet.entrySet()) { + KeyExtent ke = entry.getKey(); + String location = locations.get(ke); + + if (location == null) { + for (PathSize pathSize : entry.getValue()) { + synchronized (assignmentFailures) { + List<KeyExtent> failures = assignmentFailures.get(pathSize.path); + if (failures == null) { + failures = new ArrayList<KeyExtent>(); + assignmentFailures.put(pathSize.path, failures); + } + + failures.add(ke); + } + } + + log.warn("Could not assign " + entry.getValue().size() + " map files to tablet " + ke + " because it had no location, will retry ..."); + + continue; + } + + Map<KeyExtent,List<PathSize>> apt = assignmentsPerTabletServer.get(location); + if (apt == null) { + apt = new TreeMap<KeyExtent,List<PathSize>>(); + assignmentsPerTabletServer.put(location, apt); + } + + apt.put(entry.getKey(), entry.getValue()); + } + + ExecutorService threadPool = Executors.newFixedThreadPool(numThreads, new NamingThreadFactory("submit")); + + for (Entry<String,Map<KeyExtent,List<PathSize>>> entry : assignmentsPerTabletServer.entrySet()) { + String location = entry.getKey(); + threadPool.submit(new AssignmentTask(credentials, assignmentFailures, tableName, location, entry.getValue())); + } + + threadPool.shutdown(); + + while (!threadPool.isTerminated()) { + try { + threadPool.awaitTermination(60, TimeUnit.SECONDS); + } catch (InterruptedException e) { + e.printStackTrace(); + throw new RuntimeException(e); + } + } + + return assignmentFailures; + } + + private List<KeyExtent> assignMapFiles(Credentials credentials, String location, Map<KeyExtent,List<PathSize>> assignmentsPerTablet) + throws AccumuloException, AccumuloSecurityException { + try { + long timeInMillis = ServerConfiguration.getSystemConfiguration(instance).getTimeInMillis(Property.TSERV_BULK_TIMEOUT); + TabletClientService.Iface client = ThriftUtil.getTServerClient(location, ServerConfiguration.getSystemConfiguration(instance), timeInMillis); + try { + HashMap<KeyExtent,Map<String,org.apache.accumulo.core.data.thrift.MapFileInfo>> files = new HashMap<KeyExtent,Map<String,org.apache.accumulo.core.data.thrift.MapFileInfo>>(); + for (Entry<KeyExtent,List<PathSize>> entry : assignmentsPerTablet.entrySet()) { + HashMap<String,org.apache.accumulo.core.data.thrift.MapFileInfo> tabletFiles = new HashMap<String,org.apache.accumulo.core.data.thrift.MapFileInfo>(); + files.put(entry.getKey(), tabletFiles); + + for (PathSize pathSize : entry.getValue()) { + org.apache.accumulo.core.data.thrift.MapFileInfo mfi = new org.apache.accumulo.core.data.thrift.MapFileInfo(pathSize.estSize); + tabletFiles.put(pathSize.path.toString(), mfi); + } + } + + log.debug("Asking " + location + " to bulk load " + files); - List<TKeyExtent> failures = client.bulkImport(Tracer.traceInfo(), credentials.toThrift(instance), tid, Translator.translate(files, Translator.KET), ++ List<TKeyExtent> failures = client.bulkImport(Tracer.traceInfo(), credentials.toThrift(instance), tid, Translator.translate(files, Translators.KET), + setTime); + - return Translator.translate(failures, Translator.TKET); ++ return Translator.translate(failures, Translators.TKET); + } finally { + ThriftUtil.returnClient((TServiceClient) client); + } + } catch (ThriftSecurityException e) { + throw new AccumuloSecurityException(e.user, e.code, e); + } catch (Throwable t) { + t.printStackTrace(); + throw new AccumuloException(t); + } + } + + public static List<TabletLocation> findOverlappingTablets(AccumuloConfiguration acuConf, VolumeManager fs, TabletLocator locator, Path file, + Credentials credentials) throws Exception { + return findOverlappingTablets(acuConf, fs, locator, file, null, null, credentials); + } + + public static List<TabletLocation> findOverlappingTablets(AccumuloConfiguration acuConf, VolumeManager fs, TabletLocator locator, Path file, + KeyExtent failed, + Credentials credentials) throws Exception { + locator.invalidateCache(failed); + Text start = failed.getPrevEndRow(); + if (start != null) + start = Range.followingPrefix(start); + return findOverlappingTablets(acuConf, fs, locator, file, start, failed.getEndRow(), credentials); + } + + final static byte[] byte0 = {0}; + + public static List<TabletLocation> findOverlappingTablets(AccumuloConfiguration acuConf, VolumeManager vm, TabletLocator locator, Path file, Text startRow, + Text endRow, Credentials credentials) throws Exception { + List<TabletLocation> result = new ArrayList<TabletLocation>(); + Collection<ByteSequence> columnFamilies = Collections.emptyList(); + String filename = file.toString(); + // log.debug(filename + " finding overlapping tablets " + startRow + " -> " + endRow); + FileSystem fs = vm.getFileSystemByPath(file); + FileSKVIterator reader = FileOperations.getInstance().openReader(filename, true, fs, fs.getConf(), acuConf); + try { + Text row = startRow; + if (row == null) + row = new Text(); + while (true) { + // log.debug(filename + " Seeking to row " + row); + reader.seek(new Range(row, null), columnFamilies, false); + if (!reader.hasTop()) { + // log.debug(filename + " not found"); + break; + } + row = reader.getTopKey().getRow(); + TabletLocation tabletLocation = locator.locateTablet(credentials, row, false, true); + // log.debug(filename + " found row " + row + " at location " + tabletLocation); + result.add(tabletLocation); + row = tabletLocation.tablet_extent.getEndRow(); + if (row != null && (endRow == null || row.compareTo(endRow) < 0)) { + row = new Text(row); + row.append(byte0, 0, byte0.length); + } else + break; + } + } finally { + reader.close(); + } + // log.debug(filename + " to be sent to " + result); + return result; + } + + public static class AssignmentStats { + private Map<KeyExtent,Integer> counts; + private int numUniqueMapFiles; + private Map<Path,List<KeyExtent>> completeFailures = null; + private Set<Path> failedFailures = null; + + AssignmentStats(int fileCount) { + counts = new HashMap<KeyExtent,Integer>(); + numUniqueMapFiles = fileCount; + } + + void attemptingAssignments(Map<Path,List<TabletLocation>> assignments) { + for (Entry<Path,List<TabletLocation>> entry : assignments.entrySet()) { + for (TabletLocation tl : entry.getValue()) { + + Integer count = getCount(tl.tablet_extent); + + counts.put(tl.tablet_extent, count + 1); + } + } + } + + void assignmentsFailed(Map<Path,List<KeyExtent>> assignmentFailures) { + for (Entry<Path,List<KeyExtent>> entry : assignmentFailures.entrySet()) { + for (KeyExtent ke : entry.getValue()) { + + Integer count = getCount(ke); + + counts.put(ke, count - 1); + } + } + } + + void assignmentsAbandoned(Map<Path,List<KeyExtent>> completeFailures) { + this.completeFailures = completeFailures; + } + + void tabletSplit(KeyExtent parent, Collection<KeyExtent> children) { + Integer count = getCount(parent); + + counts.remove(parent); + + for (KeyExtent keyExtent : children) + counts.put(keyExtent, count); + } + + private Integer getCount(KeyExtent parent) { + Integer count = counts.get(parent); + + if (count == null) { + count = 0; + } + return count; + } + + void unrecoveredMapFiles(Set<Path> failedFailures) { + this.failedFailures = failedFailures; + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder(); + int totalAssignments = 0; + int tabletsImportedTo = 0; + + int min = Integer.MAX_VALUE, max = Integer.MIN_VALUE; + + for (Entry<KeyExtent,Integer> entry : counts.entrySet()) { + totalAssignments += entry.getValue(); + if (entry.getValue() > 0) + tabletsImportedTo++; + + if (entry.getValue() < min) + min = entry.getValue(); + + if (entry.getValue() > max) + max = entry.getValue(); + } + + double stddev = 0; + + for (Entry<KeyExtent,Integer> entry : counts.entrySet()) + stddev += Math.pow(entry.getValue() - totalAssignments / (double) counts.size(), 2); + + stddev = stddev / counts.size(); + stddev = Math.sqrt(stddev); + + Set<KeyExtent> failedTablets = new HashSet<KeyExtent>(); + for (List<KeyExtent> ft : completeFailures.values()) + failedTablets.addAll(ft); + + sb.append("BULK IMPORT ASSIGNMENT STATISTICS\n"); + sb.append(String.format("# of map files : %,10d%n", numUniqueMapFiles)); + sb.append(String.format("# map files with failures : %,10d %6.2f%s%n", completeFailures.size(), completeFailures.size() * 100.0 / numUniqueMapFiles, "%")); + sb.append(String.format("# failed failed map files : %,10d %s%n", failedFailures.size(), failedFailures.size() > 0 ? " <-- THIS IS BAD" : "")); + sb.append(String.format("# of tablets : %,10d%n", counts.size())); + sb.append(String.format("# tablets imported to : %,10d %6.2f%s%n", tabletsImportedTo, tabletsImportedTo * 100.0 / counts.size(), "%")); + sb.append(String.format("# tablets with failures : %,10d %6.2f%s%n", failedTablets.size(), failedTablets.size() * 100.0 / counts.size(), "%")); + sb.append(String.format("min map files per tablet : %,10d%n", min)); + sb.append(String.format("max map files per tablet : %,10d%n", max)); + sb.append(String.format("avg map files per tablet : %,10.2f (std dev = %.2f)%n", totalAssignments / (double) counts.size(), stddev)); + return sb.toString(); + } + } + +}
