| /* |
| * 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.pig.backend.hadoop.executionengine.util; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.PathFilter; |
| import org.apache.hadoop.mapreduce.InputSplit; |
| import org.apache.pig.FuncSpec; |
| import org.apache.pig.PigConfiguration; |
| import org.apache.pig.PigException; |
| import org.apache.pig.backend.executionengine.ExecException; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore; |
| import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims; |
| import org.apache.pig.data.DataBag; |
| import org.apache.pig.data.DataType; |
| import org.apache.pig.data.Tuple; |
| import org.apache.pig.data.TupleFactory; |
| import org.apache.pig.impl.PigContext; |
| import org.apache.pig.impl.builtin.PartitionSkewedKeys; |
| import org.apache.pig.impl.io.FileLocalizer; |
| import org.apache.pig.impl.io.FileSpec; |
| import org.apache.pig.impl.io.ReadToEndLoader; |
| import org.apache.pig.impl.plan.NodeIdGenerator; |
| import org.apache.pig.impl.plan.OperatorKey; |
| import org.apache.pig.impl.util.Pair; |
| import org.apache.pig.impl.util.UDFContext; |
| import org.apache.pig.impl.util.Utils; |
| |
| /** |
| * A class of utility static methods to be used in the hadoop map reduce backend |
| */ |
| public class MapRedUtil { |
| |
| private static Log log = LogFactory.getLog(MapRedUtil.class); |
| private static final TupleFactory tf = TupleFactory.getInstance(); |
| |
| public static final String FILE_SYSTEM_NAME = FileSystem.FS_DEFAULT_NAME_KEY; |
| |
| /** |
| * Loads the key distribution sampler file |
| * |
| * @param keyDistFile the name for the distribution file |
| * @param totalReducers gets set to the total number of reducers as found in the dist file |
| * @param keyType Type of the key to be stored in the return map. It currently treats Tuple as a special case. |
| */ |
| @SuppressWarnings("unchecked") |
| public static <E> Map<E, Pair<Integer, Integer>> loadPartitionFileFromLocalCache( |
| String keyDistFile, Integer[] totalReducers, byte keyType, Configuration mapConf) |
| throws IOException { |
| |
| Map<E, Pair<Integer, Integer>> reducerMap = new HashMap<E, Pair<Integer, Integer>>(); |
| |
| // use local file system to get the keyDistFile |
| Configuration conf = new Configuration(false); |
| |
| if (mapConf.get("yarn.resourcemanager.principal")!=null) { |
| conf.set("yarn.resourcemanager.principal", mapConf.get("yarn.resourcemanager.principal")); |
| } |
| |
| if (mapConf.get("fs.file.impl")!=null) |
| conf.set("fs.file.impl", mapConf.get("fs.file.impl")); |
| if (mapConf.get("fs.hdfs.impl")!=null) |
| conf.set("fs.hdfs.impl", mapConf.get("fs.hdfs.impl")); |
| |
| copyTmpFileConfigurationValues(PigMapReduce.sJobConfInternal.get(), conf); |
| |
| conf.set(MapRedUtil.FILE_SYSTEM_NAME, "file:///"); |
| |
| ReadToEndLoader loader = new ReadToEndLoader(Utils.getTmpFileStorageObject(PigMapReduce.sJobConfInternal.get()), conf, |
| keyDistFile, 0); |
| DataBag partitionList; |
| Tuple t = loader.getNext(); |
| if (t == null) { |
| // this could happen if the input directory for sampling is empty |
| log.warn("Empty dist file: " + keyDistFile); |
| return reducerMap; |
| } |
| // The keydist file is structured as (key, min, max) |
| // min, max being the index of the reducers |
| Map<String, Object > distMap = (Map<String, Object>) t.get (0); |
| partitionList = (DataBag) distMap.get(PartitionSkewedKeys.PARTITION_LIST); |
| totalReducers[0] = Integer.valueOf(""+distMap.get(PartitionSkewedKeys.TOTAL_REDUCERS)); |
| Iterator<Tuple> it = partitionList.iterator(); |
| while (it.hasNext()) { |
| Tuple idxTuple = it.next(); |
| Integer maxIndex = (Integer) idxTuple.get(idxTuple.size() - 1); |
| Integer minIndex = (Integer) idxTuple.get(idxTuple.size() - 2); |
| // Used to replace the maxIndex with the number of reducers |
| if (maxIndex < minIndex) { |
| maxIndex = totalReducers[0] + maxIndex; |
| } |
| E keyT; |
| |
| // if the join is on more than 1 key |
| if (idxTuple.size() > 3) { |
| // remove the last 2 fields of the tuple, i.e: minIndex and maxIndex and store |
| // it in the reducer map |
| Tuple keyTuple = tf.newTuple(); |
| for (int i=0; i < idxTuple.size() - 2; i++) { |
| keyTuple.append(idxTuple.get(i)); |
| } |
| keyT = (E) keyTuple; |
| } else { |
| if (keyType == DataType.TUPLE) { |
| keyT = (E)tf.newTuple(1); |
| ((Tuple)keyT).set(0,idxTuple.get(0)); |
| } else { |
| keyT = (E) idxTuple.get(0); |
| } |
| } |
| // number of reducers |
| Integer cnt = maxIndex - minIndex; |
| reducerMap.put(keyT, new Pair(minIndex, cnt));// 1 is added to account for the 0 index |
| } |
| return reducerMap; |
| } |
| |
| public static void copyTmpFileConfigurationValues(Configuration fromConf, Configuration toConf) { |
| // Currently these are used only by loaders (and not storers), so we do not need to copy |
| // mapred properties that are required by @{Link SequenceFileInterStorage} |
| |
| if (fromConf.getBoolean(PigConfiguration.PIG_ENABLE_TEMP_FILE_COMPRESSION, false)) { |
| toConf.setBoolean(PigConfiguration.PIG_ENABLE_TEMP_FILE_COMPRESSION, true); |
| if (fromConf.get(PigConfiguration.PIG_TEMP_FILE_COMPRESSION_CODEC) != null) { |
| toConf.set(PigConfiguration.PIG_TEMP_FILE_COMPRESSION_CODEC, |
| fromConf.get(PigConfiguration.PIG_TEMP_FILE_COMPRESSION_CODEC)); |
| } |
| if (fromConf.get(PigConfiguration.PIG_TEMP_FILE_COMPRESSION_STORAGE) != null) { |
| toConf.set(PigConfiguration.PIG_TEMP_FILE_COMPRESSION_STORAGE, |
| fromConf.get(PigConfiguration.PIG_TEMP_FILE_COMPRESSION_STORAGE)); |
| } |
| } |
| } |
| |
| public static void setupUDFContext(Configuration job) throws IOException { |
| UDFContext udfc = UDFContext.getUDFContext(); |
| udfc.addJobConf(job); |
| // don't deserialize in front-end |
| if (udfc.isUDFConfEmpty()) { |
| udfc.deserialize(); |
| } |
| } |
| |
| /** |
| * Sets up output and log dir paths for a single-store streaming job |
| * |
| * @param st - POStore of the current job |
| * @param pigContext |
| * @param conf |
| * @throws IOException |
| */ |
| public static void setupStreamingDirsConfSingle(POStore st, PigContext pigContext, |
| Configuration conf) throws IOException { |
| // set out filespecs |
| String outputPathString = st.getSFile().getFileName(); |
| if (HadoopShims.hasFileSystemImpl(new Path(outputPathString), conf)) { |
| conf.set("pig.streaming.log.dir", |
| new Path(outputPathString, JobControlCompiler.LOG_DIR).toString()); |
| } |
| else { |
| String tmpLocationStr = FileLocalizer.getTemporaryPath(pigContext).toString(); |
| Path tmpLocation = new Path(tmpLocationStr); |
| conf.set("pig.streaming.log.dir", |
| new Path(tmpLocation, JobControlCompiler.LOG_DIR).toString()); |
| } |
| conf.set("pig.streaming.task.output.dir", outputPathString); |
| } |
| |
| /** |
| * Sets up output and log dir paths for a multi-store streaming job |
| * |
| * @param pigContext |
| * @param conf |
| * @throws IOException |
| */ |
| public static void setupStreamingDirsConfMulti(PigContext pigContext, Configuration conf) |
| throws IOException { |
| |
| String tmpLocationStr = FileLocalizer.getTemporaryPath(pigContext).toString(); |
| Path tmpLocation = new Path(tmpLocationStr); |
| conf.set("pig.streaming.log.dir", |
| new Path(tmpLocation, JobControlCompiler.LOG_DIR).toString()); |
| conf.set("pig.streaming.task.output.dir", tmpLocation.toString()); |
| } |
| |
| public static FileSpec checkLeafIsStore( |
| PhysicalPlan plan, |
| PigContext pigContext) throws ExecException { |
| try { |
| PhysicalOperator leaf = plan.getLeaves().get(0); |
| FileSpec spec = null; |
| if(!(leaf instanceof POStore)){ |
| String scope = leaf.getOperatorKey().getScope(); |
| POStore str = new POStore(new OperatorKey(scope, |
| NodeIdGenerator.getGenerator().getNextNodeId(scope))); |
| spec = new FileSpec(FileLocalizer.getTemporaryPath( |
| pigContext).toString(), |
| new FuncSpec(Utils.getTmpFileCompressorName(pigContext))); |
| str.setSFile(spec); |
| plan.addAsLeaf(str); |
| } else{ |
| spec = ((POStore)leaf).getSFile(); |
| } |
| return spec; |
| } catch (Exception e) { |
| int errCode = 2045; |
| String msg = "Internal error. Not able to check if the leaf node is a store operator."; |
| throw new ExecException(msg, errCode, PigException.BUG, e); |
| } |
| } |
| |
| /** |
| * Get all files recursively from the given list of files |
| * |
| * @param files a list of FileStatus |
| * @param conf the configuration object |
| * @return the list of fileStatus that contains all the files in the given |
| * list and, recursively, all the files inside the directories in |
| * the given list |
| * @throws IOException |
| */ |
| public static List<FileStatus> getAllFileRecursively( |
| List<FileStatus> files, Configuration conf) throws IOException { |
| List<FileStatus> result = new ArrayList<FileStatus>(); |
| int len = files.size(); |
| for (int i = 0; i < len; ++i) { |
| FileStatus file = files.get(i); |
| if (file.isDir()) { |
| Path p = file.getPath(); |
| FileSystem fs = p.getFileSystem(conf); |
| addInputPathRecursively(result, fs, p, hiddenFileFilter); |
| } else { |
| result.add(file); |
| } |
| } |
| log.info("Total input paths to process : " + result.size()); |
| return result; |
| } |
| |
| private static void addInputPathRecursively(List<FileStatus> result, |
| FileSystem fs, Path path, PathFilter inputFilter) |
| throws IOException { |
| for (FileStatus stat: fs.listStatus(path, inputFilter)) { |
| if (stat.isDir()) { |
| addInputPathRecursively(result, fs, stat.getPath(), inputFilter); |
| } else { |
| result.add(stat); |
| } |
| } |
| } |
| |
| private static final PathFilter hiddenFileFilter = new PathFilter(){ |
| @Override |
| public boolean accept(Path p){ |
| String name = p.getName(); |
| return !name.startsWith("_") && !name.startsWith("."); |
| } |
| }; |
| |
| public static long getPathLength(FileSystem fs, FileStatus status) |
| throws IOException{ |
| return getPathLength(fs, status, Long.MAX_VALUE); |
| } |
| |
| /** |
| * Returns the total number of bytes for this file, or if a directory all |
| * files in the directory. |
| * |
| * @param fs FileSystem |
| * @param status FileStatus |
| * @param max Maximum value of total length that will trigger exit. Many |
| * times we're only interested whether the total length of files is greater |
| * than X or not. In such case, we can exit the function early as soon as |
| * the max is reached. |
| * @return |
| * @throws IOException |
| */ |
| public static long getPathLength(FileSystem fs, FileStatus status, long max) |
| throws IOException { |
| if (!status.isDir()) { |
| return status.getLen(); |
| } else { |
| FileStatus[] children = fs.listStatus( |
| status.getPath(), hiddenFileFilter); |
| long size = 0; |
| for (FileStatus child : children) { |
| size += getPathLength(fs, child, max); |
| if (size > max) return size; |
| } |
| return size; |
| } |
| } |
| |
| /* The following codes are for split combination: see PIG-1518 |
| * |
| */ |
| private static Comparator<Node> nodeComparator = new Comparator<Node>() { |
| @Override |
| public int compare(Node o1, Node o2) { |
| long cmp = o1.length - o2.length; |
| return cmp == 0 ? 0 : cmp < 0 ? -1 : 1; |
| } |
| }; |
| |
| private static final class ComparableSplit implements Comparable<ComparableSplit> { |
| private InputSplit rawInputSplit; |
| private HashSet<Node> nodes; |
| // id used as a tie-breaker when two splits are of equal size. |
| private long id; |
| ComparableSplit(InputSplit split, long id) { |
| rawInputSplit = split; |
| nodes = new HashSet<Node>(); |
| this.id = id; |
| } |
| |
| void add(Node node) { |
| nodes.add(node); |
| } |
| |
| void removeFromNodes() { |
| for (Node node : nodes) |
| node.remove(this); |
| } |
| |
| public InputSplit getSplit() { |
| return rawInputSplit; |
| } |
| |
| @Override |
| public boolean equals(Object other) { |
| if (other == null || !(other instanceof ComparableSplit)) |
| return false; |
| return (compareTo((ComparableSplit) other) == 0); |
| } |
| |
| @Override |
| public int hashCode() { |
| return 41; |
| } |
| |
| @Override |
| public int compareTo(ComparableSplit other) { |
| try { |
| long cmp = rawInputSplit.getLength() - other.rawInputSplit.getLength(); |
| // in descending order |
| return cmp == 0 ? (id == other.id ? 0 : id < other.id ? -1 : 1) : cmp < 0 ? 1 : -1; |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } catch (InterruptedException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| |
| private static class DummySplit extends InputSplit { |
| private long length; |
| |
| @Override |
| public String[] getLocations() { |
| return null; |
| } |
| |
| @Override |
| public long getLength() { |
| return length; |
| } |
| |
| public void setLength(long length) { |
| this.length = length; |
| } |
| } |
| |
| private static class Node { |
| private long length = 0; |
| private ArrayList<ComparableSplit> splits; |
| private boolean sorted; |
| |
| Node() throws IOException, InterruptedException { |
| length = 0; |
| splits = new ArrayList<ComparableSplit>(); |
| sorted = false; |
| } |
| |
| void add(ComparableSplit split) throws IOException, InterruptedException { |
| splits.add(split); |
| length++; |
| } |
| |
| void remove(ComparableSplit split) { |
| if (!sorted) |
| sort(); |
| int index = Collections.binarySearch(splits, split); |
| if (index >= 0) { |
| splits.remove(index); |
| length--; |
| } |
| } |
| |
| void sort() { |
| if (!sorted) { |
| Collections.sort(splits); |
| sorted = true; |
| } |
| } |
| |
| ArrayList<ComparableSplit> getSplits() { |
| return splits; |
| } |
| |
| public long getLength() { |
| return length; |
| } |
| } |
| |
| public static List<List<InputSplit>> getCombinePigSplits(List<InputSplit> |
| oneInputSplits, long maxCombinedSplitSize, Configuration conf) |
| throws IOException, InterruptedException { |
| ArrayList<Node> nodes = new ArrayList<Node>(); |
| HashMap<String, Node> nodeMap = new HashMap<String, Node>(); |
| List<List<InputSplit>> result = new ArrayList<List<InputSplit>>(); |
| List<Long> resultLengths = new ArrayList<Long>(); |
| long comparableSplitId = 0; |
| |
| int size = 0, nSplits = oneInputSplits.size(); |
| InputSplit lastSplit = null; |
| int emptyCnt = 0; |
| for (InputSplit split : oneInputSplits) { |
| if (split.getLength() == 0) { |
| emptyCnt++; |
| continue; |
| } |
| if (split.getLength() >= maxCombinedSplitSize) { |
| comparableSplitId++; |
| ArrayList<InputSplit> combinedSplits = new ArrayList<InputSplit>(); |
| combinedSplits.add(split); |
| result.add(combinedSplits); |
| resultLengths.add(split.getLength()); |
| } else { |
| String[] locations = split.getLocations(); |
| if (locations.length == 0) { |
| // This split is missing blocks, or the split returned bad locations. |
| // Don't try to combine. |
| comparableSplitId++; |
| ArrayList<InputSplit> combinedSplits = new ArrayList<InputSplit>(); |
| combinedSplits.add(split); |
| result.add(combinedSplits); |
| resultLengths.add(split.getLength()); |
| } else { |
| ComparableSplit csplit = new ComparableSplit(split, comparableSplitId++); |
| // sort the locations to stabilize the number of maps: PIG-1757 |
| Arrays.sort(locations); |
| HashSet<String> locationSeen = new HashSet<String>(); |
| for (String location : locations) |
| { |
| if (!locationSeen.contains(location)) |
| { |
| Node node = nodeMap.get(location); |
| if (node == null) { |
| node = new Node(); |
| nodes.add(node); |
| nodeMap.put(location, node); |
| } |
| node.add(csplit); |
| csplit.add(node); |
| locationSeen.add(location); |
| } |
| } |
| lastSplit = split; |
| size++; |
| } |
| } |
| } |
| /* verification code: debug purpose |
| { |
| ArrayList<ComparableSplit> leftoverSplits = new ArrayList<ComparableSplit>(); |
| HashSet<InputSplit> seen = new HashSet<InputSplit>(); |
| for (Node node : nodes) { |
| if (node.getLength() > 0) |
| { |
| ArrayList<ComparableSplit> splits = node.getSplits(); |
| for (ComparableSplit split : splits) { |
| if (!seen.contains(split.getSplit())) { |
| // remove duplicates. The set has to be on the raw input split not the |
| // comparable input split as the latter overrides the compareTo method |
| // so its equality semantics is changed and not we want here |
| seen.add(split.getSplit()); |
| leftoverSplits.add(split); |
| } |
| } |
| } |
| } |
| |
| int combinedSplitLen = 0; |
| for (PigSplit split : result) |
| combinedSplitLen += split.getNumPaths(); |
| if (combinedSplitLen + leftoverSplits.size()!= nSplits-emptyCnt) { |
| throw new AssertionError("number of combined splits {"+combinedSplitLen+"+"+leftoverSplits.size()+"-"+size+"} does not match the number of original splits ["+nSplits+"]."); |
| } |
| } |
| */ |
| if (nSplits > 0 && emptyCnt == nSplits) |
| { |
| // if all splits are empty, add a single empty split as currently an empty directory is |
| // not properly handled somewhere |
| ArrayList<InputSplit> combinedSplits = new ArrayList<InputSplit>(); |
| combinedSplits.add(oneInputSplits.get(0)); |
| result.add(combinedSplits); |
| } |
| else if (size == 1) { |
| ArrayList<InputSplit> combinedSplits = new ArrayList<InputSplit>(); |
| combinedSplits.add(lastSplit); |
| result.add(combinedSplits); |
| } else if (size > 1) { |
| // combine small splits |
| Collections.sort(nodes, nodeComparator); |
| DummySplit dummy = new DummySplit(); |
| // dummy is used to search for next split of suitable size to be combined |
| ComparableSplit dummyComparableSplit = new ComparableSplit(dummy, -1); |
| for (Node node : nodes) { |
| // sort the splits on this node in descending order |
| node.sort(); |
| long totalSize = 0; |
| ArrayList<ComparableSplit> splits = node.getSplits(); |
| int idx; |
| int lenSplits; |
| ArrayList<InputSplit> combinedSplits = new ArrayList<InputSplit>(); |
| ArrayList<ComparableSplit> combinedComparableSplits = new ArrayList<ComparableSplit>(); |
| while (!splits.isEmpty()) { |
| combinedSplits.add(splits.get(0).getSplit()); |
| combinedComparableSplits.add(splits.get(0)); |
| int startIdx = 1; |
| lenSplits = splits.size(); |
| totalSize += splits.get(0).getSplit().getLength(); |
| long spaceLeft = maxCombinedSplitSize - totalSize; |
| dummy.setLength(spaceLeft); |
| idx = Collections.binarySearch(node.getSplits().subList(startIdx, lenSplits), dummyComparableSplit); |
| idx = -idx-1+startIdx; |
| while (idx < lenSplits) |
| { |
| long thisLen = splits.get(idx).getSplit().getLength(); |
| combinedSplits.add(splits.get(idx).getSplit()); |
| combinedComparableSplits.add(splits.get(idx)); |
| totalSize += thisLen; |
| spaceLeft -= thisLen; |
| if (spaceLeft <= 0) |
| break; |
| // find next combinable chunk |
| startIdx = idx + 1; |
| if (startIdx >= lenSplits) |
| break; |
| dummy.setLength(spaceLeft); |
| idx = Collections.binarySearch(node.getSplits().subList(startIdx, lenSplits), dummyComparableSplit); |
| idx = -idx-1+startIdx; |
| } |
| if (totalSize > maxCombinedSplitSize/2) { |
| result.add(combinedSplits); |
| resultLengths.add(totalSize); |
| removeSplits(combinedComparableSplits); |
| totalSize = 0; |
| combinedSplits = new ArrayList<InputSplit>(); |
| combinedComparableSplits.clear(); |
| splits = node.getSplits(); |
| } else { |
| if (combinedSplits.size() != lenSplits) |
| throw new AssertionError("Combined split logic error!"); |
| break; |
| } |
| } |
| } |
| // handle leftovers |
| ArrayList<ComparableSplit> leftoverSplits = new ArrayList<ComparableSplit>(); |
| HashSet<InputSplit> seen = new HashSet<InputSplit>(); |
| for (Node node : nodes) { |
| for (ComparableSplit split : node.getSplits()) { |
| if (!seen.contains(split.getSplit())) { |
| // remove duplicates. The set has to be on the raw input split not the |
| // comparable input split as the latter overrides the compareTo method |
| // so its equality semantics is changed and not we want here |
| seen.add(split.getSplit()); |
| leftoverSplits.add(split); |
| } |
| } |
| } |
| |
| /* verification code |
| int combinedSplitLen = 0; |
| for (PigSplit split : result) |
| combinedSplitLen += split.getNumPaths(); |
| if (combinedSplitLen + leftoverSplits.size()!= nSplits-emptyCnt) |
| throw new AssertionError("number of combined splits ["+combinedSplitLen+"+"+leftoverSplits.size()+"] does not match the number of original splits ["+nSplits+"]."); |
| */ |
| if (!leftoverSplits.isEmpty()) |
| { |
| long totalSize = 0; |
| ArrayList<InputSplit> combinedSplits = new ArrayList<InputSplit>(); |
| ArrayList<ComparableSplit> combinedComparableSplits = new ArrayList<ComparableSplit>(); |
| |
| int splitLen = leftoverSplits.size(); |
| for (int i = 0; i < splitLen; i++) |
| { |
| ComparableSplit split = leftoverSplits.get(i); |
| long thisLen = split.getSplit().getLength(); |
| if (totalSize + thisLen >= maxCombinedSplitSize) { |
| removeSplits(combinedComparableSplits); |
| result.add(combinedSplits); |
| resultLengths.add(totalSize); |
| combinedSplits = new ArrayList<InputSplit>(); |
| combinedComparableSplits.clear(); |
| totalSize = 0; |
| } |
| combinedSplits.add(split.getSplit()); |
| combinedComparableSplits.add(split); |
| totalSize += split.getSplit().getLength(); |
| if (i == splitLen - 1) { |
| // last piece: it could be very small, try to see it can be squeezed into any existing splits |
| for (int j =0; j < result.size(); j++) |
| { |
| if (resultLengths.get(j) + totalSize <= maxCombinedSplitSize) |
| { |
| List<InputSplit> isList = result.get(j); |
| for (InputSplit csplit : combinedSplits) { |
| isList.add(csplit); |
| } |
| removeSplits(combinedComparableSplits); |
| combinedSplits.clear(); |
| break; |
| } |
| } |
| if (!combinedSplits.isEmpty()) { |
| // last piece can not be squeezed in, create a new combined split for them. |
| removeSplits(combinedComparableSplits); |
| result.add(combinedSplits); |
| } |
| } |
| } |
| } |
| } |
| /* verification codes |
| int combinedSplitLen = 0; |
| for (PigSplit split : result) |
| combinedSplitLen += split.getNumPaths(); |
| if (combinedSplitLen != nSplits-emptyCnt) |
| throw new AssertionError("number of combined splits ["+combinedSplitLen+"] does not match the number of original splits ["+nSplits+"]."); |
| |
| long totalLen = 0; |
| for (PigSplit split : result) |
| totalLen += split.getLength(); |
| |
| long origTotalLen = 0; |
| for (InputSplit split : oneInputSplits) |
| origTotalLen += split.getLength(); |
| if (totalLen != origTotalLen) |
| throw new AssertionError("The total length ["+totalLen+"] does not match the original ["+origTotalLen+"]"); |
| */ |
| log.info("Total input paths (combined) to process : " + result.size()); |
| return result; |
| } |
| |
| private static void removeSplits(List<ComparableSplit> splits) { |
| for (ComparableSplit split: splits) |
| split.removeFromNodes(); |
| } |
| |
| public String inputSplitToString(InputSplit[] splits) throws IOException, InterruptedException { |
| // debugging purpose only |
| StringBuilder st = new StringBuilder(); |
| st.append("Number of splits :" + splits.length+"\n"); |
| long len = 0; |
| for (InputSplit split: splits) |
| len += split.getLength(); |
| st.append("Total Length = "+ len +"\n"); |
| for (int i = 0; i < splits.length; i++) { |
| st.append("Input split["+i+"]:\n Length = "+ splits[i].getLength()+"\n Locations:\n"); |
| for (String location : splits[i].getLocations()) |
| st.append(" "+location+"\n"); |
| st.append("\n-----------------------\n"); |
| } |
| return st.toString(); |
| } |
| |
| /* verification code: debug purpose only |
| public String inputSplitToString(ArrayList<ComparableSplit> splits) throws IOException, InterruptedException { |
| StringBuilder st = new StringBuilder(); |
| st.append("Number of splits :" + splits.size()+"\n"); |
| long len = 0; |
| for (ComparableSplit split: splits) |
| len += split.getSplit().getLength(); |
| st.append("Total Length = "+ len +"\n"); |
| for (int i = 0; i < splits.size(); i++) { |
| st.append("Input split["+i+"]:\n Length = "+ splits.get(i).getSplit().getLength()+"\n Locations:\n"); |
| for (String location : splits.get(i).getSplit().getLocations()) |
| st.append(" "+location+"\n"); |
| st.append("\n-----------------------\n"); |
| } |
| return st.toString(); |
| } |
| */ |
| } |
