| /** |
| * 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.tez; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.TreeMap; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.io.RawComparator; |
| import org.apache.hadoop.io.WritableComparable; |
| import org.apache.hadoop.io.WritableComparator; |
| import org.apache.hadoop.mapred.JobConf; |
| import org.apache.hadoop.mapreduce.InputFormat; |
| import org.apache.hadoop.mapreduce.Job; |
| import org.apache.hadoop.mapreduce.filecache.ClientDistributedCacheManager; |
| import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat; |
| import org.apache.hadoop.mapreduce.v2.util.MRApps; |
| import org.apache.hadoop.yarn.api.records.LocalResource; |
| import org.apache.hadoop.yarn.api.records.LocalResourceType; |
| import org.apache.hadoop.yarn.api.records.LocalResourceVisibility; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.util.ConverterUtils; |
| import org.apache.pig.PigConfiguration; |
| import org.apache.pig.PigException; |
| import org.apache.pig.StoreFuncInterface; |
| import org.apache.pig.backend.executionengine.ExecException; |
| import org.apache.pig.backend.hadoop.DateTimeWritable; |
| import org.apache.pig.backend.hadoop.HDataType; |
| import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil; |
| import org.apache.pig.backend.hadoop.executionengine.JobCreationException; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.DistinctCombiner; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.InputSizeReducerEstimator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigSecondaryKeyGroupComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PhyPlanSetter; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigBigDecimalRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigBigIntegerRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigBooleanRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigBytesRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigCombiner; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigDateTimeRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigDoubleRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigFloatRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigInputFormat; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigIntRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigLongRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigOutputFormat; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSecondaryKeyComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigTextRawComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigTupleSortComparator; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigWritableComparators; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.partitioners.SecondaryKeyPartitioner; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.EndOfAllInputSetter; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSplit; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezEdgeDescriptor; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezOpPlanVisitor; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezOperPlan; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezOperator; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.TezPOPackageAnnotator.LoRearrangeDiscoverer; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POLocalRearrangeTez; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POShuffleTezLoad; |
| import org.apache.pig.backend.hadoop.executionengine.tez.plan.operator.POStoreTez; |
| import org.apache.pig.backend.hadoop.executionengine.tez.runtime.PartitionerDefinedVertexManager; |
| import org.apache.pig.backend.hadoop.executionengine.tez.runtime.PigGraceShuffleVertexManager; |
| import org.apache.pig.backend.hadoop.executionengine.tez.runtime.PigInputFormatTez; |
| import org.apache.pig.backend.hadoop.executionengine.tez.runtime.PigOutputFormatTez; |
| import org.apache.pig.backend.hadoop.executionengine.tez.runtime.PigProcessor; |
| import org.apache.pig.backend.hadoop.executionengine.tez.util.MRToTezHelper; |
| import org.apache.pig.backend.hadoop.executionengine.tez.util.SecurityHelper; |
| import org.apache.pig.backend.hadoop.executionengine.tez.util.TezCompilerUtil; |
| import org.apache.pig.backend.hadoop.executionengine.tez.util.TezInputHelper; |
| import org.apache.pig.backend.hadoop.executionengine.tez.util.TezUDFContextSeparator; |
| import org.apache.pig.data.DataType; |
| import org.apache.pig.impl.PigContext; |
| import org.apache.pig.impl.PigImplConstants; |
| import org.apache.pig.impl.io.FileLocalizer; |
| import org.apache.pig.impl.io.NullablePartitionWritable; |
| import org.apache.pig.impl.io.NullableTuple; |
| import org.apache.pig.impl.plan.DependencyOrderWalker; |
| import org.apache.pig.impl.plan.OperatorKey; |
| import org.apache.pig.impl.plan.VisitorException; |
| import org.apache.pig.impl.util.ObjectSerializer; |
| import org.apache.pig.impl.util.Pair; |
| import org.apache.pig.impl.util.UDFContext; |
| import org.apache.pig.impl.util.UDFContextSeparator.UDFType; |
| import org.apache.pig.tools.pigstats.tez.TezScriptState; |
| import org.apache.pig.tools.pigstats.tez.TezScriptState.TezDAGScriptInfo; |
| import org.apache.tez.common.TezUtils; |
| import org.apache.tez.dag.api.DAG; |
| import org.apache.tez.dag.api.DataSinkDescriptor; |
| import org.apache.tez.dag.api.DataSourceDescriptor; |
| import org.apache.tez.dag.api.Edge; |
| import org.apache.tez.dag.api.EdgeManagerPluginDescriptor; |
| import org.apache.tez.dag.api.EdgeProperty; |
| import org.apache.tez.dag.api.EdgeProperty.DataMovementType; |
| import org.apache.tez.dag.api.GroupInputEdge; |
| import org.apache.tez.dag.api.InputDescriptor; |
| import org.apache.tez.dag.api.InputInitializerDescriptor; |
| import org.apache.tez.dag.api.OutputCommitterDescriptor; |
| import org.apache.tez.dag.api.OutputDescriptor; |
| import org.apache.tez.dag.api.ProcessorDescriptor; |
| import org.apache.tez.dag.api.TezConfiguration; |
| import org.apache.tez.dag.api.UserPayload; |
| import org.apache.tez.dag.api.Vertex; |
| import org.apache.tez.dag.api.VertexGroup; |
| import org.apache.tez.dag.api.VertexLocationHint; |
| import org.apache.tez.dag.api.VertexManagerPluginDescriptor; |
| import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager; |
| import org.apache.tez.mapreduce.combine.MRCombiner; |
| import org.apache.tez.mapreduce.committer.MROutputCommitter; |
| import org.apache.tez.mapreduce.common.MRInputSplitDistributor; |
| import org.apache.tez.mapreduce.hadoop.InputSplitInfo; |
| import org.apache.tez.mapreduce.hadoop.InputSplitInfoMem; |
| import org.apache.tez.mapreduce.hadoop.MRHelpers; |
| import org.apache.tez.mapreduce.hadoop.MRJobConfig; |
| import org.apache.tez.mapreduce.input.MRInput; |
| import org.apache.tez.mapreduce.output.MROutput; |
| import org.apache.tez.mapreduce.partition.MRPartitioner; |
| import org.apache.tez.mapreduce.protos.MRRuntimeProtos; |
| import org.apache.tez.mapreduce.protos.MRRuntimeProtos.MRInputUserPayloadProto.Builder; |
| import org.apache.tez.mapreduce.protos.MRRuntimeProtos.MRSplitsProto; |
| import org.apache.tez.runtime.library.api.TezRuntimeConfiguration; |
| import org.apache.tez.runtime.library.input.ConcatenatedMergedKeyValueInput; |
| import org.apache.tez.runtime.library.input.OrderedGroupedKVInput; |
| import org.apache.tez.runtime.library.input.OrderedGroupedMergedKVInput; |
| import org.apache.tez.runtime.library.input.UnorderedKVInput; |
| |
| /** |
| * A visitor to construct DAG out of Tez plan. |
| */ |
| public class TezDagBuilder extends TezOpPlanVisitor { |
| private static final Log log = LogFactory.getLog(TezDagBuilder.class); |
| |
| private static long SHUFFLE_BYTES_PER_REDUCER_GROUPBY_DEFAULT = 384 * 1024 * 1024L; |
| private static long SHUFFLE_BYTES_PER_REDUCER_DEFAULT = 256 * 1024 * 1024L; |
| |
| private DAG dag; |
| private Map<String, LocalResource> localResources; |
| private PigContext pc; |
| private Configuration globalConf; |
| private Configuration pigContextConf; |
| private Configuration shuffleVertexManagerBaseConf; |
| private FileSystem fs; |
| private long intermediateTaskInputSize; |
| private Set<String> inputSplitInDiskVertices; |
| private TezUDFContextSeparator udfContextSeparator; |
| |
| private String serializedTezPlan; |
| private String serializedPigContext; |
| private String serializedUDFImportList; |
| |
| // Map corresponds to root vertices, reduce to intermediate and leaf vertices |
| private Resource mapTaskResource; |
| private Resource reduceTaskResource; |
| private Map<String, String> mapTaskEnv = new HashMap<String, String>(); |
| private Map<String, String> reduceTaskEnv = new HashMap<String, String>(); |
| private String mapTaskLaunchCmdOpts; |
| private String reduceTaskLaunchCmdOpts; |
| |
| private boolean disableDAGRecovery = false; |
| |
| public TezDagBuilder(PigContext pc, TezOperPlan plan, DAG dag, |
| Map<String, LocalResource> localResources) { |
| super(plan, new DependencyOrderWalker<TezOperator, TezOperPlan>(plan)); |
| this.pc = pc; |
| this.localResources = localResources; |
| this.dag = dag; |
| this.inputSplitInDiskVertices = new HashSet<String>(); |
| |
| try { |
| initialize(pc); |
| |
| udfContextSeparator = new TezUDFContextSeparator(plan, |
| new DependencyOrderWalker<TezOperator, TezOperPlan>(plan)); |
| udfContextSeparator.visit(); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| public boolean shouldDisableDAGRecovery() { |
| return disableDAGRecovery; |
| } |
| |
| private void initialize(PigContext pc) throws IOException { |
| |
| this.globalConf = ConfigurationUtil.toConfiguration(pc.getProperties(), true); |
| |
| this.pigContextConf = ConfigurationUtil.toConfiguration(pc.getProperties(), false); |
| MRToTezHelper.processMRSettings(pigContextConf, globalConf); |
| |
| shuffleVertexManagerBaseConf = new Configuration(false); |
| // Only copy tez.shuffle-vertex-manager config to keep payload size small |
| Iterator<Entry<String, String>> iter = pigContextConf.iterator(); |
| while (iter.hasNext()) { |
| Entry<String, String> entry = iter.next(); |
| if (entry.getKey().startsWith("tez.shuffle-vertex-manager")) { |
| shuffleVertexManagerBaseConf.set(entry.getKey(), entry.getValue()); |
| } |
| } |
| |
| // Add credentials from binary token file and get tokens for namenodes |
| // specified in mapreduce.job.hdfs-servers |
| SecurityHelper.populateTokenCache(globalConf, dag.getCredentials()); |
| |
| // All these classes are @InterfaceAudience.Private in Hadoop. Switch to Tez methods in TEZ-1012 |
| // set the timestamps, public/private visibility of the archives and files |
| ClientDistributedCacheManager |
| .determineTimestampsAndCacheVisibilities(globalConf); |
| // get DelegationToken for each cached file |
| ClientDistributedCacheManager.getDelegationTokens(globalConf, |
| dag.getCredentials()); |
| MRApps.setupDistributedCache(globalConf, this.localResources); |
| dag.addTaskLocalFiles(this.localResources); |
| |
| int mapMemoryMB; |
| int reduceMemoryMB; |
| int mapVCores; |
| int reduceVCores; |
| if (globalConf.get(TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB) != null) { |
| mapMemoryMB = globalConf.getInt( |
| TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB, |
| TezConfiguration.TEZ_TASK_RESOURCE_MEMORY_MB_DEFAULT); |
| reduceMemoryMB = mapMemoryMB; |
| } else { |
| // If tez setting is not defined, try MR setting |
| mapMemoryMB = globalConf.getInt(MRJobConfig.MAP_MEMORY_MB, |
| MRJobConfig.DEFAULT_MAP_MEMORY_MB); |
| reduceMemoryMB = globalConf.getInt(MRJobConfig.REDUCE_MEMORY_MB, |
| MRJobConfig.DEFAULT_REDUCE_MEMORY_MB); |
| } |
| |
| if (globalConf.get(TezConfiguration.TEZ_TASK_RESOURCE_CPU_VCORES) != null) { |
| mapVCores = globalConf.getInt( |
| TezConfiguration.TEZ_TASK_RESOURCE_CPU_VCORES, |
| TezConfiguration.TEZ_TASK_RESOURCE_CPU_VCORES_DEFAULT); |
| reduceVCores = mapVCores; |
| } else { |
| mapVCores = globalConf.getInt(MRJobConfig.MAP_CPU_VCORES, |
| MRJobConfig.DEFAULT_MAP_CPU_VCORES); |
| reduceVCores = globalConf.getInt(MRJobConfig.REDUCE_CPU_VCORES, |
| MRJobConfig.DEFAULT_REDUCE_CPU_VCORES); |
| } |
| mapTaskResource = Resource.newInstance(mapMemoryMB, mapVCores); |
| reduceTaskResource = Resource.newInstance(reduceMemoryMB, reduceVCores); |
| |
| if (globalConf.get(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS) == null) { |
| // If tez setting is not defined |
| MRHelpers.updateEnvBasedOnMRTaskEnv(globalConf, mapTaskEnv, true); |
| MRHelpers.updateEnvBasedOnMRTaskEnv(globalConf, reduceTaskEnv, false); |
| } |
| |
| if (globalConf.get(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS) != null) { |
| mapTaskLaunchCmdOpts = globalConf.get(TezConfiguration.TEZ_TASK_LAUNCH_CMD_OPTS); |
| reduceTaskLaunchCmdOpts = mapTaskLaunchCmdOpts; |
| } else { |
| // If tez setting is not defined, try MR setting |
| mapTaskLaunchCmdOpts = MRHelpers.getJavaOptsForMRMapper(globalConf); |
| reduceTaskLaunchCmdOpts = MRHelpers.getJavaOptsForMRReducer(globalConf); |
| } |
| |
| try { |
| fs = FileSystem.get(globalConf); |
| intermediateTaskInputSize = fs.getDefaultBlockSize(FileLocalizer.getTemporaryResourcePath(pc)); |
| } catch (Exception e) { |
| log.warn("Unable to get the block size for temporary directory, defaulting to 128MB", e); |
| intermediateTaskInputSize = 134217728L; |
| } |
| // At least 128MB. Else we will end up with too many tasks |
| intermediateTaskInputSize = Math.max(intermediateTaskInputSize, 134217728L); |
| intermediateTaskInputSize = Math.min(intermediateTaskInputSize, |
| globalConf.getLong( |
| InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM, |
| InputSizeReducerEstimator.DEFAULT_BYTES_PER_REDUCER)); |
| |
| serializedPigContext = ObjectSerializer.serialize(pc); |
| serializedUDFImportList = ObjectSerializer.serialize(PigContext.getPackageImportList()); |
| } |
| |
| private String getSerializedTezPlan() throws IOException { |
| if (serializedTezPlan == null) { |
| // Initialize lazy instead of constructor as this might not be needed |
| serializedTezPlan = ObjectSerializer.serialize(getPlan()); |
| } |
| return serializedTezPlan; |
| } |
| |
| // Hack to turn off relocalization till TEZ-2192 is fixed. |
| public void avoidContainerReuseIfInputSplitInDisk() throws IOException { |
| if (!inputSplitInDiskVertices.isEmpty()) { |
| // Create empty job.split file and add as resource to all other |
| // vertices that are not reading splits from disk so that their |
| // containers are not reused by vertices that read splits from disk |
| Path jobSplitFile = new Path(FileLocalizer.getTemporaryPath(pc), |
| MRJobConfig.JOB_SPLIT); |
| FSDataOutputStream out = fs.create(jobSplitFile); |
| out.close(); |
| log.info("Creating empty job.split in " + jobSplitFile); |
| FileStatus splitFileStatus = fs.getFileStatus(jobSplitFile); |
| LocalResource localResource = LocalResource.newInstance( |
| ConverterUtils.getYarnUrlFromPath(jobSplitFile), |
| LocalResourceType.FILE, |
| LocalResourceVisibility.APPLICATION, |
| splitFileStatus.getLen(), |
| splitFileStatus.getModificationTime()); |
| for (Vertex vertex : dag.getVertices()) { |
| if (!inputSplitInDiskVertices.contains(vertex.getName())) { |
| if (vertex.getTaskLocalFiles().containsKey( |
| MRJobConfig.JOB_SPLIT)) { |
| throw new RuntimeException( |
| "LocalResources already contains a" |
| + " resource named " |
| + MRJobConfig.JOB_SPLIT); |
| } |
| vertex.getTaskLocalFiles().put(MRJobConfig.JOB_SPLIT, |
| localResource); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public void visitTezOp(TezOperator tezOp) throws VisitorException { |
| TezOperPlan tezPlan = getPlan(); |
| List<TezOperator> predecessors = tezPlan.getPredecessors(tezOp); |
| // Construct vertex for the current Tez operator |
| Vertex to = null; |
| try { |
| if (!tezOp.isVertexGroup()) { |
| to = newVertex(tezOp); |
| dag.addVertex(to); |
| } else { |
| // For union, we construct VertexGroup after iterating the |
| // predecessors. |
| } |
| } catch (Exception e) { |
| throw new VisitorException("Cannot create vertex for " |
| + tezOp.name(), e); |
| } |
| |
| // Connect the new vertex with predecessor vertices |
| if (predecessors != null) { |
| Vertex[] groupMembers = new Vertex[predecessors.size()]; |
| |
| for (int i = 0; i < predecessors.size(); i++) { |
| // Since this is a dependency order walker, predecessor vertices |
| // must have already been created. |
| TezOperator pred = predecessors.get(i); |
| try { |
| if (pred.isVertexGroup()) { |
| VertexGroup from = pred.getVertexGroupInfo().getVertexGroup(); |
| // The plan of vertex group is empty. Since we create the Edge based on |
| // some of the operators in the plan refer to one of the vertex group members. |
| // Both the vertex group and its members reference same EdgeDescriptor object to the |
| // the successor |
| GroupInputEdge edge = newGroupInputEdge( |
| getPlan().getOperator(pred.getVertexGroupMembers().get(0)), tezOp, from, to); |
| dag.addEdge(edge); |
| } else { |
| Vertex from = dag.getVertex(pred.getOperatorKey().toString()); |
| if (tezOp.isVertexGroup()) { |
| groupMembers[i] = from; |
| } else { |
| EdgeProperty prop = newEdge(pred, tezOp, false); |
| Edge edge = Edge.create(from, to, prop); |
| dag.addEdge(edge); |
| } |
| } |
| } catch (IOException e) { |
| throw new VisitorException("Cannot create edge from " |
| + pred.name() + " to " + tezOp.name(), e); |
| } |
| } |
| |
| if (tezOp.isVertexGroup()) { |
| String groupName = tezOp.getOperatorKey().toString(); |
| VertexGroup vertexGroup = dag.createVertexGroup(groupName, groupMembers); |
| tezOp.getVertexGroupInfo().setVertexGroup(vertexGroup); |
| POStore store = tezOp.getVertexGroupInfo().getStore(); |
| if (store != null) { |
| String outputKey = store.getOperatorKey().toString(); |
| if (store instanceof POStoreTez) { |
| outputKey = ((POStoreTez) store).getOutputKey(); |
| } |
| vertexGroup.addDataSink(outputKey, |
| DataSinkDescriptor.create(tezOp.getVertexGroupInfo().getStoreOutputDescriptor(), |
| OutputCommitterDescriptor.create(MROutputCommitter.class.getName()), dag.getCredentials())); |
| } |
| } |
| } |
| } |
| |
| private GroupInputEdge newGroupInputEdge(TezOperator fromOp, |
| TezOperator toOp, VertexGroup from, Vertex to) throws IOException { |
| |
| EdgeProperty edgeProperty = newEdge(fromOp, toOp, true); |
| |
| String groupInputClass = ConcatenatedMergedKeyValueInput.class.getName(); |
| |
| // In case of SCATTER_GATHER and UnorderedKVInput it will still be |
| // ConcatenatedMergedKeyValueInput |
| if(edgeProperty.getDataMovementType().equals(DataMovementType.SCATTER_GATHER) |
| && edgeProperty.getEdgeDestination().getClassName().equals(OrderedGroupedKVInput.class.getName())) { |
| groupInputClass = OrderedGroupedMergedKVInput.class.getName(); |
| } |
| |
| return GroupInputEdge.create(from, to, edgeProperty, |
| InputDescriptor.create(groupInputClass).setUserPayload(edgeProperty.getEdgeDestination().getUserPayload())); |
| } |
| |
| /** |
| * Return EdgeProperty that connects two vertices. |
| * |
| * @param from |
| * @param to |
| * @return EdgeProperty |
| * @throws IOException |
| */ |
| private EdgeProperty newEdge(TezOperator from, TezOperator to, boolean isMergedInput) |
| throws IOException { |
| TezEdgeDescriptor edge = to.inEdges.get(from.getOperatorKey()); |
| PhysicalPlan combinePlan = edge.combinePlan; |
| |
| InputDescriptor in = InputDescriptor.create(edge.inputClassName); |
| OutputDescriptor out = OutputDescriptor.create(edge.outputClassName); |
| |
| Configuration conf = new Configuration(pigContextConf); |
| |
| if (edge.needsDistinctCombiner()) { |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS, |
| MRCombiner.class.getName()); |
| conf.set(MRJobConfig.COMBINE_CLASS_ATTR, |
| DistinctCombiner.Combine.class.getName()); |
| log.info("Setting distinct combiner class between " |
| + from.getOperatorKey() + " and " + to.getOperatorKey()); |
| } else if (!combinePlan.isEmpty()) { |
| udfContextSeparator.serializeUDFContextForEdge(conf, from, to, UDFType.USERFUNC); |
| addCombiner(combinePlan, to, conf, isMergedInput); |
| } |
| |
| List<POLocalRearrangeTez> lrs = PlanHelper.getPhysicalOperators(from.plan, |
| POLocalRearrangeTez.class); |
| |
| for (POLocalRearrangeTez lr : lrs) { |
| if (lr.containsOutputKey(to.getOperatorKey().toString())) { |
| byte keyType = lr.getKeyType(); |
| setIntermediateOutputKeyValue(keyType, conf, to, lr.isConnectedToPackage(), isMergedInput); |
| // In case of secondary key sort, main key type is the actual key type |
| conf.set("pig.reduce.key.type", Byte.toString(lr.getMainKeyType())); |
| break; |
| } |
| } |
| |
| conf.setIfUnset(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, |
| MRPartitioner.class.getName()); |
| |
| if (edge.getIntermediateOutputKeyClass() != null) { |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, |
| edge.getIntermediateOutputKeyClass()); |
| } |
| |
| if (edge.getIntermediateOutputValueClass() != null) { |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, |
| edge.getIntermediateOutputValueClass()); |
| } |
| |
| if (edge.getIntermediateOutputKeyComparatorClass() != null) { |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| edge.getIntermediateOutputKeyComparatorClass()); |
| } |
| |
| conf.setBoolean(MRConfiguration.MAPPER_NEW_API, true); |
| conf.setBoolean(MRConfiguration.REDUCER_NEW_API, true); |
| conf.setBoolean(PigImplConstants.PIG_EXECTYPE_MODE_LOCAL, pc.getExecType().isLocal()); |
| conf.set(PigImplConstants.PIG_LOG4J_PROPERTIES, ObjectSerializer.serialize(pc.getLog4jProperties())); |
| conf.set("udf.import.list", serializedUDFImportList); |
| |
| if(to.isGlobalSort() || to.isLimitAfterSort()){ |
| conf.set("pig.sortOrder", |
| ObjectSerializer.serialize(to.getSortOrder())); |
| } |
| |
| if (edge.isUseSecondaryKey()) { |
| conf.set("pig.secondarySortOrder", |
| ObjectSerializer.serialize(edge.getSecondarySortOrder())); |
| conf.set(org.apache.hadoop.mapreduce.MRJobConfig.PARTITIONER_CLASS_ATTR, |
| SecondaryKeyPartitioner.class.getName()); |
| // These needs to be on the vertex as well for POShuffleTezLoad to pick it up. |
| // Tez framework also expects this to be per vertex and not edge. IFile.java picks |
| // up keyClass and valueClass from vertex config. TODO - check with Tez folks |
| // In MR - job.setSortComparatorClass() or MRJobConfig.KEY_COMPARATOR |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| PigSecondaryKeyComparator.class.getName()); |
| // In MR - job.setOutputKeyClass() or MRJobConfig.OUTPUT_KEY_CLASS |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, NullableTuple.class.getName()); |
| setGroupingComparator(conf, PigSecondaryKeyGroupComparator.class.getName()); |
| } |
| |
| if (edge.partitionerClass != null) { |
| conf.set(org.apache.hadoop.mapreduce.MRJobConfig.PARTITIONER_CLASS_ATTR, |
| edge.partitionerClass.getName()); |
| } |
| |
| UserPayload payLoad = TezUtils.createUserPayloadFromConf(conf); |
| out.setUserPayload(payLoad); |
| in.setUserPayload(payLoad); |
| |
| // Remove combiner and reset payload |
| if (!combinePlan.isEmpty()) { |
| boolean noCombineInReducer = false; |
| boolean noCombineInMapper = edge.getCombinerInMap() == null ? false : !edge.getCombinerInMap(); |
| String reducerNoCombiner = globalConf.get(PigConfiguration.PIG_EXEC_NO_COMBINER_REDUCER); |
| if (edge.getCombinerInReducer() != null) { |
| noCombineInReducer = !edge.getCombinerInReducer(); |
| } else if (reducerNoCombiner == null || reducerNoCombiner.equals("auto")) { |
| noCombineInReducer = TezCompilerUtil.bagDataTypeInCombinePlan(combinePlan); |
| } else { |
| noCombineInReducer = Boolean.parseBoolean(reducerNoCombiner); |
| } |
| if (noCombineInReducer || noCombineInMapper) { |
| log.info("Turning off combiner in reducer vertex " + to.getOperatorKey() + " for edge from " + from.getOperatorKey()); |
| conf.unset(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS); |
| conf.unset(MRJobConfig.COMBINE_CLASS_ATTR); |
| conf.unset("pig.combinePlan"); |
| conf.unset("pig.combine.package"); |
| conf.unset("pig.map.keytype"); |
| UserPayload payLoadWithoutCombiner = TezUtils.createUserPayloadFromConf(conf); |
| if (noCombineInMapper) { |
| out.setUserPayload(payLoadWithoutCombiner); |
| } |
| if (noCombineInReducer) { |
| in.setUserPayload(payLoadWithoutCombiner); |
| } |
| } |
| } |
| |
| if (edge.dataMovementType!=DataMovementType.BROADCAST && to.getEstimatedParallelism()!=-1 && to.getVertexParallelism()==-1 && (to.isGlobalSort()||to.isSkewedJoin())) { |
| // Use custom edge |
| return EdgeProperty.create((EdgeManagerPluginDescriptor)null, |
| edge.dataSourceType, edge.schedulingType, out, in); |
| } |
| |
| if (to.isUseGraceParallelism()) { |
| // Put datamovement to null to prevent vertex "to" from starting. It will be started by PigGraceShuffleVertexManager |
| return EdgeProperty.create((EdgeManagerPluginDescriptor)null, edge.dataSourceType, |
| edge.schedulingType, out, in); |
| } |
| return EdgeProperty.create(edge.dataMovementType, edge.dataSourceType, |
| edge.schedulingType, out, in); |
| } |
| |
| private void addCombiner(PhysicalPlan combinePlan, TezOperator pkgTezOp, |
| Configuration conf, boolean isMergedInput) throws IOException { |
| POPackage combPack = (POPackage) combinePlan.getRoots().get(0); |
| POLocalRearrange combRearrange = (POLocalRearrange) combinePlan |
| .getLeaves().get(0); |
| setIntermediateOutputKeyValue(combRearrange.getKeyType(), conf, pkgTezOp, true, isMergedInput); |
| |
| LoRearrangeDiscoverer lrDiscoverer = new LoRearrangeDiscoverer( |
| combinePlan, null, pkgTezOp, combPack); |
| lrDiscoverer.visit(); |
| |
| combinePlan.remove(combPack); |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_COMBINER_CLASS, |
| MRCombiner.class.getName()); |
| conf.set(MRJobConfig.COMBINE_CLASS_ATTR, |
| PigCombiner.Combine.class.getName()); |
| conf.set("pig.combinePlan", ObjectSerializer.serialize(combinePlan)); |
| conf.set("pig.combine.package", ObjectSerializer.serialize(combPack)); |
| conf.set("pig.map.keytype", ObjectSerializer |
| .serialize(new byte[] { combRearrange.getKeyType() })); |
| } |
| |
| private Vertex newVertex(TezOperator tezOp) throws IOException, |
| ClassNotFoundException, InterruptedException { |
| |
| ProcessorDescriptor procDesc = ProcessorDescriptor.create( |
| tezOp.getProcessorName()); |
| |
| // Pass physical plans to vertex as user payload. |
| JobConf payloadConf = new JobConf(pigContextConf); |
| |
| // We do this so that dag.getCredentials(), job.getCredentials(), |
| // job.getConfiguration().getCredentials() all reference the same Credentials object |
| // Unfortunately there is no setCredentials() on Job |
| payloadConf.setCredentials(dag.getCredentials()); |
| // We won't actually use this job, but we need it to talk with the Load Store funcs |
| @SuppressWarnings("deprecation") |
| Job job = new Job(payloadConf); |
| payloadConf = (JobConf) job.getConfiguration(); |
| //TODO: Investigate. Setting as map writes empty output. |
| //payloadConf.setBoolean(MRConfig.IS_MAP_PROCESSOR, tezOp.isUseMRMapSettings()); |
| payloadConf.setBoolean(MRConfiguration.MAPPER_NEW_API, true); |
| payloadConf.setBoolean(MRConfiguration.REDUCER_NEW_API, true); |
| payloadConf.setClass(MRConfiguration.INPUTFORMAT_CLASS, |
| PigInputFormatTez.class, InputFormat.class); |
| setOutputFormat(job); |
| payloadConf.set("udf.import.list", serializedUDFImportList); |
| payloadConf.set("exectype", "TEZ"); |
| payloadConf.setBoolean(PigImplConstants.PIG_EXECTYPE_MODE_LOCAL, pc.getExecType().isLocal()); |
| payloadConf.set(PigImplConstants.PIG_LOG4J_PROPERTIES, ObjectSerializer.serialize(pc.getLog4jProperties())); |
| |
| DateTimeWritable.setupAvailableZoneIds(); |
| |
| // Process stores |
| LinkedList<POStore> stores = processStores(tezOp, payloadConf, job); |
| |
| // Process UserFuncs |
| processUserFuncs(tezOp, job); |
| |
| Configuration inputPayLoad = null; |
| Configuration outputPayLoad = null; |
| |
| if (!stores.isEmpty()) { |
| outputPayLoad = new Configuration(payloadConf); |
| outputPayLoad.set(JobControlCompiler.PIG_MAP_STORES, |
| ObjectSerializer.serialize(new ArrayList<POStore>())); |
| } |
| |
| if (!(tezOp.getLoaderInfo().getLoads().isEmpty())) { |
| payloadConf.set(PigInputFormat.PIG_LOADS, ObjectSerializer.serialize(tezOp.getLoaderInfo().getLoads())); |
| payloadConf.set(PigInputFormat.PIG_INPUT_SIGNATURES, ObjectSerializer.serialize(tezOp.getLoaderInfo().getInpSignatureLists())); |
| payloadConf.set(PigInputFormat.PIG_INPUT_LIMITS, ObjectSerializer.serialize(tezOp.getLoaderInfo().getInpLimits())); |
| inputPayLoad = new Configuration(payloadConf); |
| } |
| |
| if (tezOp.getSampleOperator() != null) { |
| payloadConf.set(PigProcessor.SAMPLE_VERTEX, tezOp.getSampleOperator().getOperatorKey().toString()); |
| } |
| |
| if (tezOp.getSortOperator() != null) { |
| // Required by Sample Aggregation job for estimating quantiles |
| payloadConf.set(PigProcessor.SORT_VERTEX, tezOp.getSortOperator().getOperatorKey().toString()); |
| // PIG-4162: Order by/Skew Join in intermediate stage. |
| // Increasing order by parallelism may not be required as it is |
| // usually followed by limit other than store. But would benefit |
| // cases like skewed join followed by group by. |
| if (tezOp.getSortOperator().getEstimatedParallelism() != -1 |
| && tezOp.getSortOperator().isIntermediateReducer()) { |
| payloadConf.setLong( |
| InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM, |
| intermediateTaskInputSize); |
| } |
| |
| } |
| |
| // Set parent plan for all operators in the Tez plan. |
| new PhyPlanSetter(tezOp.plan).visit(); |
| |
| // Set the endOfAllInput flag on the physical plan if certain operators that |
| // use this property (such as STREAM) are present in the plan. |
| EndOfAllInputSetter.EndOfAllInputChecker checker = |
| new EndOfAllInputSetter.EndOfAllInputChecker(tezOp.plan); |
| checker.visit(); |
| if (checker.isEndOfAllInputPresent()) { |
| payloadConf.set(JobControlCompiler.END_OF_INP_IN_MAP, "true"); |
| } |
| |
| // Configure the classes for incoming shuffles to this TezOp |
| // TODO: Refactor out resetting input keys, PIG-3957 |
| List<PhysicalOperator> roots = tezOp.plan.getRoots(); |
| if (roots.size() == 1 && roots.get(0) instanceof POPackage) { |
| POPackage pack = (POPackage) roots.get(0); |
| |
| List<PhysicalOperator> succsList = tezOp.plan.getSuccessors(pack); |
| if (succsList != null) { |
| succsList = new ArrayList<PhysicalOperator>(succsList); |
| } |
| byte keyType = pack.getPkgr().getKeyType(); |
| tezOp.plan.remove(pack); |
| payloadConf.set("pig.reduce.package", ObjectSerializer.serialize(pack)); |
| |
| POShuffleTezLoad newPack = new POShuffleTezLoad(pack); |
| if (tezOp.isSkewedJoin()) { |
| newPack.setSkewedJoins(true); |
| } |
| tezOp.plan.add(newPack); |
| |
| boolean isMergedInput = false; |
| // Set input keys for POShuffleTezLoad. This is used to identify |
| // the inputs that are attached to the POShuffleTezLoad in the |
| // backend. |
| Map<Integer, String> localRearrangeMap = new TreeMap<Integer, String>(); |
| TezOperator from = null; |
| for (TezOperator pred : mPlan.getPredecessors(tezOp)) { |
| if (tezOp.getSampleOperator() != null && tezOp.getSampleOperator() == pred) { |
| // skip sample vertex input |
| } else { |
| String inputKey = pred.getOperatorKey().toString(); |
| boolean isVertexGroup = false; |
| if (pred.isVertexGroup()) { |
| isVertexGroup = true; |
| pred = mPlan.getOperator(pred.getVertexGroupMembers().get(0)); |
| } |
| LinkedList<POLocalRearrangeTez> lrs = |
| PlanHelper.getPhysicalOperators(pred.plan, POLocalRearrangeTez.class); |
| for (POLocalRearrangeTez lr : lrs) { |
| if (lr.isConnectedToPackage() |
| && lr.containsOutputKey(tezOp.getOperatorKey().toString())) { |
| localRearrangeMap.put((int) lr.getIndex(), inputKey); |
| if (isVertexGroup) { |
| isMergedInput = true; |
| } |
| from = pred; |
| } |
| } |
| } |
| } |
| for (Map.Entry<Integer, String> entry : localRearrangeMap.entrySet()) { |
| newPack.addInputKey(entry.getValue()); |
| } |
| |
| if (succsList != null) { |
| for (PhysicalOperator succs : succsList) { |
| tezOp.plan.connect(newPack, succs); |
| } |
| } |
| |
| //POShuffleTezLoad accesses the comparator setting |
| selectKeyComparator(keyType, payloadConf, tezOp, isMergedInput); |
| |
| if (tezOp.isUseSecondaryKey()) { |
| TezEdgeDescriptor edge = tezOp.inEdges.get(from.getOperatorKey()); |
| // Currently only PigSecondaryKeyGroupingComparator is used in POShuffleTezLoad. |
| // When PIG-4685: SecondaryKeyOptimizerTez does not optimize cogroup is fixed |
| // in future, PigSecondaryKeyComparator will have to be used and that will require this. |
| payloadConf.set("pig.secondarySortOrder", ObjectSerializer |
| .serialize(edge.getSecondarySortOrder())); |
| } |
| |
| } |
| |
| // set parent plan in all operators. currently the parent plan is really |
| // used only when POStream, POSplit are present in the plan |
| new PhyPlanSetter(tezOp.plan).visit(); |
| |
| // Serialize the execution plan |
| payloadConf.set(PigProcessor.PLAN, |
| ObjectSerializer.serialize(tezOp.plan)); |
| |
| udfContextSeparator.serializeUDFContext(payloadConf, tezOp); |
| |
| if (!pc.inIllustrator) { |
| for (POStore store : stores) { |
| // unset inputs for POStore, otherwise, map/reduce plan will be unnecessarily deserialized |
| store.setInputs(null); |
| store.setParentPlan(null); |
| } |
| // We put them in the reduce because PigOutputCommitter checks the |
| // ID of the task to see if it's a map, and if not, calls the reduce |
| // committers. |
| payloadConf.set(JobControlCompiler.PIG_MAP_STORES, |
| ObjectSerializer.serialize(new ArrayList<POStore>())); |
| payloadConf.set(JobControlCompiler.PIG_REDUCE_STORES, |
| ObjectSerializer.serialize(stores)); |
| } |
| |
| if (tezOp.isNeedEstimateParallelism()) { |
| payloadConf.setBoolean(PigProcessor.ESTIMATE_PARALLELISM, true); |
| log.info("Estimate quantile for sample aggregation vertex " + tezOp.getOperatorKey().toString()); |
| } |
| |
| // set various parallelism into the job conf for later analysis, PIG-2779 |
| payloadConf.setInt(PigImplConstants.REDUCER_DEFAULT_PARALLELISM, pc.defaultParallel); |
| payloadConf.setInt(PigImplConstants.REDUCER_REQUESTED_PARALLELISM, tezOp.getRequestedParallelism()); |
| payloadConf.setInt(PigImplConstants.REDUCER_ESTIMATED_PARALLELISM, tezOp.getEstimatedParallelism()); |
| |
| TezScriptState ss = TezScriptState.get(); |
| ss.addVertexSettingsToConf(dag.getName(), tezOp, payloadConf); |
| |
| // Take our assembled configuration and create a vertex |
| UserPayload userPayload = TezUtils.createUserPayloadFromConf(payloadConf); |
| TezDAGScriptInfo dagScriptInfo = TezScriptState.get().getDAGScriptInfo(dag.getName()); |
| String alias = dagScriptInfo.getAlias(tezOp); |
| String aliasLocation = dagScriptInfo.getAliasLocation(tezOp); |
| String features = dagScriptInfo.getPigFeatures(tezOp); |
| String vertexInfo = aliasLocation + " (" + features + ")" ; |
| procDesc.setUserPayload(userPayload).setHistoryText(TezUtils.convertToHistoryText(vertexInfo, payloadConf)); |
| |
| String vmPluginName = null; |
| Configuration vmPluginConf = null; |
| boolean containScatterGather = false; |
| boolean containCustomPartitioner = false; |
| for (TezEdgeDescriptor edge : tezOp.inEdges.values()) { |
| if (edge.dataMovementType == DataMovementType.SCATTER_GATHER) { |
| containScatterGather = true; |
| } |
| if (edge.partitionerClass != null) { |
| containCustomPartitioner = true; |
| } |
| } |
| |
| if(containScatterGather) { |
| vmPluginName = ShuffleVertexManager.class.getName(); |
| vmPluginConf = new Configuration(shuffleVertexManagerBaseConf); |
| } |
| |
| // Set the right VertexManagerPlugin |
| if (tezOp.getEstimatedParallelism() != -1) { |
| boolean autoParallelism = false; |
| if (tezOp.isGlobalSort()||tezOp.isSkewedJoin()) { |
| if (tezOp.getVertexParallelism()==-1 && ( |
| tezOp.isGlobalSort() &&getPlan().getPredecessors(tezOp).size()==1|| |
| tezOp.isSkewedJoin() &&getPlan().getPredecessors(tezOp).size()==2)) { |
| // Set VertexManagerPlugin to PartitionerDefinedVertexManager, which is able |
| // to decrease/increase parallelism of sorting vertex dynamically |
| // based on the numQuantiles calculated by sample aggregation vertex |
| vmPluginName = PartitionerDefinedVertexManager.class.getName(); |
| autoParallelism = true; |
| log.info("Set VertexManagerPlugin to PartitionerDefinedParallelismVertexManager for vertex " + tezOp.getOperatorKey().toString()); |
| } |
| } else { |
| if (containScatterGather && !containCustomPartitioner) { |
| |
| // For Intermediate reduce, set the bytes per reducer to be block size. |
| long bytesPerReducer = intermediateTaskInputSize; |
| // If there are store statements, use BYTES_PER_REDUCER_PARAM configured by user. |
| // If not as default use 384MB for group bys and 256 MB for joins. Not using |
| // default 1G as that value was suited for mapreduce logic where numReducers=(map input size/bytesPerReducer). |
| // In Tez, numReducers=(map output size/bytesPerReducer) we need lower values to avoid skews in reduce |
| // as map input sizes are mostly always high compared to map output. |
| if (stores.size() > 0) { |
| if (pigContextConf.get(InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM) != null) { |
| bytesPerReducer = pigContextConf.getLong( |
| InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM, |
| InputSizeReducerEstimator.DEFAULT_BYTES_PER_REDUCER); |
| } else if (tezOp.isGroupBy()) { |
| bytesPerReducer = SHUFFLE_BYTES_PER_REDUCER_GROUPBY_DEFAULT; |
| } else { |
| bytesPerReducer = SHUFFLE_BYTES_PER_REDUCER_DEFAULT; |
| } |
| } |
| |
| // Use auto-parallelism feature of ShuffleVertexManager to dynamically |
| // reduce the parallelism of the vertex. Use PigGraceShuffleVertexManager |
| // instead of ShuffleVertexManager if pig.tez.grace.parallelism is turned on |
| if (payloadConf.getBoolean(PigConfiguration.PIG_TEZ_GRACE_PARALLELISM, true) |
| && !TezOperPlan.getGrandParentsForGraceParallelism(getPlan(), tezOp).isEmpty() |
| && tezOp.getCrossKeys() == null) { |
| vmPluginName = PigGraceShuffleVertexManager.class.getName(); |
| tezOp.setUseGraceParallelism(true); |
| vmPluginConf.set("pig.tez.plan", getSerializedTezPlan()); |
| vmPluginConf.set(PigImplConstants.PIG_CONTEXT, serializedPigContext); |
| vmPluginConf.setLong(InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM, bytesPerReducer); |
| } |
| vmPluginConf.setBoolean(ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_ENABLE_AUTO_PARALLEL, true); |
| vmPluginConf.setLong(ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_DESIRED_TASK_INPUT_SIZE, bytesPerReducer); |
| autoParallelism = true; |
| log.info("Set auto parallelism for vertex " + tezOp.getOperatorKey().toString()); |
| } |
| } |
| if (globalConf.getBoolean(PigConfiguration.PIG_TEZ_AUTO_PARALLELISM_DISABLE_DAG_RECOVERY, false) && autoParallelism) { |
| disableDAGRecovery = true; |
| } |
| } |
| if (tezOp.isLimit() && (vmPluginName == null || vmPluginName.equals(PigGraceShuffleVertexManager.class.getName())|| |
| vmPluginName.equals(ShuffleVertexManager.class.getName()))) { |
| if (tezOp.inEdges.values().iterator().next().inputClassName.equals(UnorderedKVInput.class.getName())) { |
| // Setting SRC_FRACTION to 0.00001 so that even if there are 100K source tasks, |
| // limit job starts when 1 source task finishes. |
| // If limit is part of a group by or join because their parallelism is 1, |
| // we should leave the configuration with the defaults. |
| vmPluginConf = (vmPluginConf == null) ? new Configuration(pigContextConf) : vmPluginConf; |
| vmPluginConf.set(ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MIN_SRC_FRACTION, "0.00001"); |
| vmPluginConf.set(ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MAX_SRC_FRACTION, "0.00001"); |
| log.info("Set " + ShuffleVertexManager.TEZ_SHUFFLE_VERTEX_MANAGER_MIN_SRC_FRACTION + " to 0.00001 for limit vertex " + tezOp.getOperatorKey().toString()); |
| } |
| } |
| |
| int parallel = tezOp.getVertexParallelism(); |
| if (tezOp.isUseGraceParallelism()) { |
| parallel = -1; |
| } |
| Resource resource = tezOp.isUseMRMapSettings() ? mapTaskResource : reduceTaskResource; |
| |
| Vertex vertex = Vertex.create(tezOp.getOperatorKey().toString(), procDesc, parallel, resource); |
| |
| if (tezOp.isUseMRMapSettings()) { |
| vertex.setTaskLaunchCmdOpts(mapTaskLaunchCmdOpts); |
| vertex.setTaskEnvironment(mapTaskEnv); |
| } else { |
| vertex.setTaskLaunchCmdOpts(reduceTaskLaunchCmdOpts); |
| vertex.setTaskEnvironment(reduceTaskEnv); |
| } |
| |
| MRToTezHelper.setVertexConfig(vertex, tezOp.isUseMRMapSettings(), globalConf); |
| |
| log.info("For vertex - " + tezOp.getOperatorKey().toString() |
| + ": parallelism=" + tezOp.getVertexParallelism() |
| + ", memory=" + vertex.getTaskResource().getMemory() |
| + ", java opts=" + vertex.getTaskLaunchCmdOpts() |
| ); |
| log.info("Processing aliases: " + alias); |
| log.info("Detailed locations: " + aliasLocation); |
| log.info("Pig features in the vertex: " + features); |
| // Right now there can only be one of each of these. Will need to be |
| // more generic when there can be more. |
| for (POLoad ld : tezOp.getLoaderInfo().getLoads()) { |
| |
| // TODO: These should get the globalConf, or a merged version that |
| // keeps settings like pig.maxCombinedSplitSize |
| Builder userPayLoadBuilder = MRRuntimeProtos.MRInputUserPayloadProto.newBuilder(); |
| |
| InputSplitInfo inputSplitInfo = tezOp.getLoaderInfo().getInputSplitInfo(); |
| Map<String, LocalResource> additionalLocalResources = null; |
| int spillThreshold = payloadConf |
| .getInt(PigConfiguration.PIG_TEZ_INPUT_SPLITS_MEM_THRESHOLD, |
| PigConfiguration.PIG_TEZ_INPUT_SPLITS_MEM_THRESHOLD_DEFAULT); |
| |
| // Currently inputSplitInfo is always InputSplitInfoMem at this point |
| if (inputSplitInfo instanceof InputSplitInfoMem) { |
| MRSplitsProto.Builder splitsBuilder = MRSplitsProto.newBuilder(); |
| Pair<Long, Boolean> serializationInfo = TezInputHelper.createSplitsProto(inputSplitInfo, pigContextConf, splitsBuilder, |
| spillThreshold); |
| MRSplitsProto splitsProto = splitsBuilder.build(); |
| if(!serializationInfo.second) { |
| //write to disk |
| inputPayLoad.setBoolean( |
| org.apache.tez.mapreduce.hadoop.MRJobConfig.MR_TEZ_SPLITS_VIA_EVENTS, |
| false); |
| // Write splits to disk |
| Path inputSplitsDir = FileLocalizer.getTemporaryPath(pc); |
| log.info("Writing input splits to " + inputSplitsDir |
| + " for vertex " + vertex.getName() |
| + " as the partially serialized size in memory is " |
| + serializationInfo.first + ". Configured " |
| + PigConfiguration.PIG_TEZ_INPUT_SPLITS_MEM_THRESHOLD |
| + " is " + spillThreshold); |
| inputSplitInfo = MRToTezHelper.writeInputSplitInfoToDisk( |
| (InputSplitInfoMem)inputSplitInfo, inputSplitsDir, payloadConf, fs, splitsProto); |
| additionalLocalResources = new HashMap<String, LocalResource>(); |
| MRToTezHelper.updateLocalResourcesForInputSplits( |
| fs, inputSplitInfo, |
| additionalLocalResources); |
| inputSplitInDiskVertices.add(vertex.getName()); |
| } else { |
| // Send splits via RPC to AM |
| userPayLoadBuilder.setSplits(splitsProto); |
| } |
| //Free up memory |
| tezOp.getLoaderInfo().setInputSplitInfo(null); |
| } |
| |
| udfContextSeparator.serializeUDFContext(inputPayLoad, tezOp, UDFType.LOADFUNC); |
| userPayLoadBuilder.setConfigurationBytes(TezUtils.createByteStringFromConf(inputPayLoad)); |
| |
| vertex.setLocationHint(VertexLocationHint.create(inputSplitInfo.getTaskLocationHints())); |
| vertex.addDataSource(ld.getOperatorKey().toString(), |
| DataSourceDescriptor.create(InputDescriptor.create(MRInput.class.getName()) |
| .setUserPayload(UserPayload.create(userPayLoadBuilder.build().toByteString().asReadOnlyByteBuffer())), |
| InputInitializerDescriptor.create(MRInputSplitDistributor.class.getName()), |
| inputSplitInfo.getNumTasks(), |
| dag.getCredentials(), |
| null, |
| additionalLocalResources)); |
| } |
| |
| // Union within a split can have multiple stores writing to same output |
| Set<String> uniqueStoreOutputs = new HashSet<String>(); |
| for (POStore store : stores) { |
| |
| ArrayList<POStore> singleStore = new ArrayList<POStore>(); |
| singleStore.add(store); |
| |
| Configuration outPayLoad = new Configuration(outputPayLoad); |
| udfContextSeparator.serializeUDFContext(outPayLoad, tezOp, store); |
| outPayLoad.set(JobControlCompiler.PIG_REDUCE_STORES, |
| ObjectSerializer.serialize(singleStore)); |
| |
| OutputDescriptor storeOutDescriptor = OutputDescriptor.create( |
| MROutput.class.getName()).setUserPayload(TezUtils |
| .createUserPayloadFromConf(outPayLoad)); |
| if (tezOp.getVertexGroupStores() != null) { |
| OperatorKey vertexGroupKey = tezOp.getVertexGroupStores().get(store.getOperatorKey()); |
| if (vertexGroupKey != null) { |
| getPlan().getOperator(vertexGroupKey).getVertexGroupInfo() |
| .setStoreOutputDescriptor(storeOutDescriptor); |
| continue; |
| } |
| } |
| String outputKey = ((POStoreTez) store).getOutputKey(); |
| if (!uniqueStoreOutputs.contains(outputKey)) { |
| vertex.addDataSink(outputKey.toString(), |
| DataSinkDescriptor.create(storeOutDescriptor, |
| OutputCommitterDescriptor.create(MROutputCommitter.class.getName()), |
| dag.getCredentials())); |
| uniqueStoreOutputs.add(outputKey); |
| } |
| } |
| |
| // LoadFunc and StoreFunc add delegation tokens to Job Credentials in |
| // setLocation and setStoreLocation respectively. For eg: HBaseStorage |
| // InputFormat add delegation token in getSplits and OutputFormat in |
| // checkOutputSpecs. For eg: FileInputFormat and FileOutputFormat |
| if (stores.size() > 0) { |
| new PigOutputFormat().checkOutputSpecs(job); |
| } |
| |
| // else if(tezOp.isLimitAfterSort()) |
| // TODO: PIG-4049 If standalone Limit we need a new VertexManager or new input |
| // instead of ShuffledMergedInput. For limit part of the sort (order by parallel 1) itself |
| // need to enhance PartitionerDefinedVertexManager |
| |
| if (vmPluginName != null) { |
| VertexManagerPluginDescriptor vmPluginDescriptor = VertexManagerPluginDescriptor.create(vmPluginName); |
| if (vmPluginConf != null) { |
| vmPluginDescriptor.setUserPayload(TezUtils.createUserPayloadFromConf(vmPluginConf)); |
| } |
| vertex.setVertexManagerPlugin(vmPluginDescriptor); |
| } |
| // Reset udfcontext jobconf. It is not supposed to be set in the front end |
| UDFContext.getUDFContext().addJobConf(null); |
| return vertex; |
| } |
| |
| /** |
| * Process POUserFunc to add credentials |
| * @param tezOp |
| * @param job |
| * @throws VisitorException |
| */ |
| private void processUserFuncs(TezOperator tezOp, Job job) throws VisitorException { |
| List<POUserFunc> userFuncs = PlanHelper.getPhysicalOperators(tezOp.plan, POUserFunc.class); |
| for (POUserFunc userFunc : userFuncs) { |
| userFunc.getFunc().addCredentials(job.getCredentials(), job.getConfiguration()); |
| } |
| } |
| |
| private LinkedList<POStore> processStores(TezOperator tezOp, |
| Configuration payloadConf, Job job) throws VisitorException, |
| IOException { |
| LinkedList<POStore> stores = PlanHelper.getPhysicalOperators( |
| tezOp.plan, POStore.class); |
| |
| if (stores.size() > 0) { |
| |
| ArrayList<POStore> storeLocations = new ArrayList<POStore>(); |
| for (POStore st : stores) { |
| storeLocations.add(st); |
| StoreFuncInterface sFunc = st.getStoreFunc(); |
| sFunc.setStoreLocation(st.getSFile().getFileName(), job); |
| sFunc.addCredentials(job.getCredentials(), job.getConfiguration()); |
| } |
| |
| Path tmpLocation = null; |
| if (stores.size() == 1) { |
| POStore st = stores.get(0); |
| |
| // set out filespecs |
| String outputPathString = st.getSFile().getFileName(); |
| if (!outputPathString.contains("://") |
| || outputPathString.startsWith("hdfs://")) { |
| payloadConf.set("pig.streaming.log.dir", new Path( |
| outputPathString, JobControlCompiler.LOG_DIR) |
| .toString()); |
| } else { |
| String tmpLocationStr = FileLocalizer.getTemporaryPath(pc) |
| .toString(); |
| tmpLocation = new Path(tmpLocationStr); |
| payloadConf.set("pig.streaming.log.dir", new Path(tmpLocation, |
| JobControlCompiler.LOG_DIR).toString()); |
| } |
| payloadConf.set("pig.streaming.task.output.dir", outputPathString); |
| |
| if(tezOp.plan.getLeaves().get(0) instanceof POSplit) { |
| // Set this so that we get correct counters |
| st.setMultiStore(true); |
| } |
| |
| } else { // multi store case |
| log.info("Setting up multi store job"); |
| String tmpLocationStr = FileLocalizer.getTemporaryPath(pc) |
| .toString(); |
| tmpLocation = new Path(tmpLocationStr); |
| |
| boolean disableCounter = payloadConf.getBoolean( |
| "pig.disable.counter", false); |
| if (disableCounter) { |
| log.info("Disable Pig custom output counters"); |
| } |
| int idx = 0; |
| for (POStore sto : storeLocations) { |
| sto.setDisableCounter(disableCounter); |
| sto.setMultiStore(true); |
| sto.setIndex(idx++); |
| } |
| |
| payloadConf.set("pig.streaming.log.dir", new Path(tmpLocation, |
| JobControlCompiler.LOG_DIR).toString()); |
| payloadConf.set("pig.streaming.task.output.dir", |
| tmpLocation.toString()); |
| } |
| |
| } |
| return stores; |
| } |
| |
| @SuppressWarnings("rawtypes") |
| private void setIntermediateOutputKeyValue(byte keyType, Configuration conf, TezOperator tezOp, |
| boolean isConnectedToPackage, boolean isMergedInput) throws JobCreationException, ExecException { |
| if (tezOp != null && tezOp.isUseSecondaryKey() && isConnectedToPackage) { |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, |
| NullableTuple.class.getName()); |
| } else if (tezOp != null && tezOp.isSkewedJoin() && isConnectedToPackage) { |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, |
| NullablePartitionWritable.class.getName()); |
| } else { |
| Class<? extends WritableComparable> keyClass = HDataType |
| .getWritableComparableTypes(keyType).getClass(); |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, |
| keyClass.getName()); |
| |
| } |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, |
| NullableTuple.class.getName()); |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, |
| MRPartitioner.class.getName()); |
| selectKeyComparator(keyType, conf, tezOp, isMergedInput); |
| } |
| |
| private static Class<? extends WritableComparator> getRawBytesComparator( |
| byte keyType) throws JobCreationException { |
| |
| // These comparators only compare bytes and we will use them except for |
| // order by for faster sorting. |
| // This ordering is good enough to be fed to reducer (POShuffleTezLoad) |
| // which will use the full comparator (GroupingComparator) for correct |
| // sorting and grouping. |
| // TODO: PIG-4652. Till Tez exposes a way to get bytes of keys being compared, |
| // we can use this only for groupby and distinct which are single inputs in |
| // POShuffleTezLoad and not join which has multiple inputs. |
| |
| switch (keyType) { |
| case DataType.BOOLEAN: |
| return PigWritableComparators.PigBooleanRawBytesComparator.class; |
| |
| case DataType.INTEGER: |
| return PigWritableComparators.PigIntRawBytesComparator.class; |
| |
| case DataType.BIGINTEGER: |
| return PigWritableComparators.PigBigIntegerRawBytesComparator.class; |
| |
| case DataType.BIGDECIMAL: |
| return PigWritableComparators.PigBigDecimalRawBytesComparator.class; |
| |
| case DataType.LONG: |
| return PigWritableComparators.PigLongRawBytesComparator.class; |
| |
| case DataType.FLOAT: |
| return PigWritableComparators.PigFloatRawBytesComparator.class; |
| |
| case DataType.DOUBLE: |
| return PigWritableComparators.PigDoubleRawBytesComparator.class; |
| |
| case DataType.DATETIME: |
| return PigWritableComparators.PigDateTimeRawBytesComparator.class; |
| |
| case DataType.CHARARRAY: |
| return PigWritableComparators.PigTextRawBytesComparator.class; |
| |
| case DataType.BYTEARRAY: |
| return PigWritableComparators.PigBytesRawBytesComparator.class; |
| |
| case DataType.MAP: |
| int errCode = 1068; |
| String msg = "Using Map as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| case DataType.TUPLE: |
| return PigWritableComparators.PigTupleSortBytesComparator.class; |
| |
| case DataType.BAG: |
| errCode = 1068; |
| msg = "Using Bag as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| default: |
| errCode = 2036; |
| msg = "Unhandled key type " + DataType.findTypeName(keyType); |
| throw new JobCreationException(msg, errCode, PigException.BUG); |
| } |
| } |
| |
| private static Class<? extends WritableComparator> getRawComparator(byte keyType) |
| throws JobCreationException { |
| |
| // These are full comparators used in order by jobs and as GroupingComparator in |
| // POShuffleTezLoad for other operations. |
| |
| // Mapreduce uses PigGrouping<DataType>WritableComparator for non-orderby jobs. |
| // In Tez, we will use the raw comparators itself on the reduce side as well as it is |
| // now fixed to handle nulls for different indexes. |
| // Also PigGrouping<DataType>WritableComparator use PigNullablePartitionWritable.compareTo |
| // which is not that efficient for cases like tuple where tuple is iterated for null checking |
| // instead of taking advantage of TupleRawComparator.hasComparedTupleNull(). |
| // Also skips multi-query index checking |
| |
| switch (keyType) { |
| case DataType.BOOLEAN: |
| return PigBooleanRawComparator.class; |
| |
| case DataType.INTEGER: |
| return PigIntRawComparator.class; |
| |
| case DataType.BIGINTEGER: |
| return PigBigIntegerRawComparator.class; |
| |
| case DataType.BIGDECIMAL: |
| return PigBigDecimalRawComparator.class; |
| |
| case DataType.LONG: |
| return PigLongRawComparator.class; |
| |
| case DataType.FLOAT: |
| return PigFloatRawComparator.class; |
| |
| case DataType.DOUBLE: |
| return PigDoubleRawComparator.class; |
| |
| case DataType.DATETIME: |
| return PigDateTimeRawComparator.class; |
| |
| case DataType.CHARARRAY: |
| return PigTextRawComparator.class; |
| |
| case DataType.BYTEARRAY: |
| return PigBytesRawComparator.class; |
| |
| case DataType.MAP: |
| int errCode = 1068; |
| String msg = "Using Map as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| case DataType.TUPLE: |
| return PigTupleSortComparator.class; |
| |
| case DataType.BAG: |
| errCode = 1068; |
| msg = "Using Bag as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| default: |
| errCode = 2036; |
| msg = "Unhandled key type " + DataType.findTypeName(keyType); |
| throw new JobCreationException(msg, errCode, PigException.BUG); |
| } |
| } |
| |
| private static Class<? extends WritableComparator> getRawBytesComparatorForSkewedJoin(byte keyType) |
| throws JobCreationException { |
| |
| // Extended Raw Bytes Comparators for SkewedJoin which unwrap the NullablePartitionWritable |
| switch (keyType) { |
| case DataType.BOOLEAN: |
| return PigWritableComparators.PigBooleanRawBytesPartitionComparator.class; |
| |
| case DataType.INTEGER: |
| return PigWritableComparators.PigIntRawBytesPartitionComparator.class; |
| |
| case DataType.BIGINTEGER: |
| return PigWritableComparators.PigBigIntegerRawBytesPartitionComparator.class; |
| |
| case DataType.BIGDECIMAL: |
| return PigWritableComparators.PigBigDecimalRawBytesPartitionComparator.class; |
| |
| case DataType.LONG: |
| return PigWritableComparators.PigLongRawBytesPartitionComparator.class; |
| |
| case DataType.FLOAT: |
| return PigWritableComparators.PigFloatRawBytesPartitionComparator.class; |
| |
| case DataType.DOUBLE: |
| return PigWritableComparators.PigDoubleRawBytesPartitionComparator.class; |
| |
| case DataType.DATETIME: |
| return PigWritableComparators.PigDateTimeRawBytesPartitionComparator.class; |
| |
| case DataType.CHARARRAY: |
| return PigWritableComparators.PigTextRawBytesPartitionComparator.class; |
| |
| case DataType.BYTEARRAY: |
| return PigWritableComparators.PigBytesRawBytesPartitionComparator.class; |
| |
| case DataType.MAP: |
| int errCode = 1068; |
| String msg = "Using Map as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| case DataType.TUPLE: |
| return PigWritableComparators.PigTupleSortBytesPartitionComparator.class; |
| |
| case DataType.BAG: |
| errCode = 1068; |
| msg = "Using Bag as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| default: |
| errCode = 2036; |
| msg = "Unhandled key type " + DataType.findTypeName(keyType); |
| throw new JobCreationException(msg, errCode, PigException.BUG); |
| } |
| } |
| |
| private static Class<? extends WritableComparator> getRawComparatorForSkewedJoin(byte keyType) |
| throws JobCreationException { |
| |
| // Extended Raw Comparators for SkewedJoin which unwrap the NullablePartitionWritable |
| switch (keyType) { |
| case DataType.BOOLEAN: |
| return PigWritableComparators.PigBooleanRawPartitionComparator.class; |
| |
| case DataType.INTEGER: |
| return PigWritableComparators.PigIntRawPartitionComparator.class; |
| |
| case DataType.BIGINTEGER: |
| return PigWritableComparators.PigBigIntegerRawPartitionComparator.class; |
| |
| case DataType.BIGDECIMAL: |
| return PigWritableComparators.PigBigDecimalRawPartitionComparator.class; |
| |
| case DataType.LONG: |
| return PigWritableComparators.PigLongRawPartitionComparator.class; |
| |
| case DataType.FLOAT: |
| return PigWritableComparators.PigFloatRawPartitionComparator.class; |
| |
| case DataType.DOUBLE: |
| return PigWritableComparators.PigDoubleRawPartitionComparator.class; |
| |
| case DataType.DATETIME: |
| return PigWritableComparators.PigDateTimeRawPartitionComparator.class; |
| |
| case DataType.CHARARRAY: |
| return PigWritableComparators.PigTextRawPartitionComparator.class; |
| |
| case DataType.BYTEARRAY: |
| return PigWritableComparators.PigBytesRawPartitionComparator.class; |
| |
| case DataType.MAP: |
| int errCode = 1068; |
| String msg = "Using Map as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| case DataType.TUPLE: |
| return PigWritableComparators.PigTupleSortPartitionComparator.class; |
| |
| case DataType.BAG: |
| errCode = 1068; |
| msg = "Using Bag as key not supported."; |
| throw new JobCreationException(msg, errCode, PigException.INPUT); |
| |
| default: |
| errCode = 2036; |
| msg = "Unhandled key type " + DataType.findTypeName(keyType); |
| throw new JobCreationException(msg, errCode, PigException.BUG); |
| } |
| } |
| |
| void selectKeyComparator(byte keyType, Configuration conf, TezOperator tezOp, boolean isMergedInput) |
| throws JobCreationException { |
| // TODO: Handle sorting like in JobControlCompiler |
| // TODO: Group comparators as in JobControlCompiler |
| if (tezOp == null) { |
| return; |
| } |
| if (tezOp.isUseSecondaryKey()) { |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| PigSecondaryKeyComparator.class.getName()); |
| setGroupingComparator(conf, PigSecondaryKeyGroupComparator.class.getName()); |
| } else { |
| // If it is not a merged input (OrderedGroupedMergedKVInput) from union then |
| // use bytes only comparator. This is temporary till PIG-4652 is done |
| if (!isMergedInput && (tezOp.isGroupBy() || tezOp.isDistinct())) { |
| conf.setClass( |
| TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| getRawBytesComparator(keyType), RawComparator.class); |
| } else if (tezOp.isSkewedJoin()) { |
| conf.setClass(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| getRawComparatorForSkewedJoin(keyType), RawComparator.class); |
| } else { |
| conf.setClass( |
| TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| getRawComparator(keyType), RawComparator.class); |
| } |
| |
| // Comparators now |
| // groupby/distinct : Comparator - RawBytesComparator |
| // groupby/distinct after union : Comparator - RawComparator |
| // orderby : Comparator - RawComparator |
| // skewed join : Comparator - RawPartitionComparator |
| // Rest (other joins) : Comparator - RawComparator |
| |
| //TODO: In PIG-4652: After Tez support for exposing key bytes |
| // groupby/distinct : Comparator - RawBytesComparator. No grouping comparator required. |
| // orderby : Comparator - RawComparator. No grouping comparator required. |
| // skewed join : Comparator - RawBytesPartitionComparator, GroupingComparator - RawPartitionComparator |
| // Rest (other joins) : Comparator - RawBytesComparator, GroupingComparator - RawComparator |
| |
| /* |
| if (tezOp.isSkewedJoin()) { |
| conf.setClass(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| getRawBytesComparatorForSkewedJoin(keyType), RawComparator.class); |
| setGroupingComparator(conf, getRawComparatorForSkewedJoin(keyType).getName()); |
| } else if (tezOp.isGroupBy() || tezOp.isDistinct()) { |
| conf.setClass( |
| TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| getRawBytesComparator(keyType), RawComparator.class); |
| } else if (hasOrderby(tezOp)) { |
| conf.setClass( |
| TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| getRawComparator(keyType), RawComparator.class); |
| } else { |
| conf.setClass( |
| TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS, |
| getRawBytesComparator(keyType), RawComparator.class); |
| setGroupingComparator(conf, getRawComparator(keyType).getName()); |
| } |
| */ |
| } |
| } |
| |
| private boolean hasOrderby(TezOperator tezOp) { |
| boolean hasOrderBy = tezOp.isGlobalSort() || tezOp.isLimitAfterSort(); |
| if (!hasOrderBy) { |
| // Check if it is a Orderby sampler job |
| List<TezOperator> succs = getPlan().getSuccessors(tezOp); |
| if (succs != null && succs.size() == 1) { |
| if (succs.get(0).isGlobalSort()) { |
| hasOrderBy = true; |
| } |
| } |
| } |
| return hasOrderBy; |
| } |
| |
| |
| private void setGroupingComparator(Configuration conf, String comparatorClass) { |
| // In MR - job.setGroupingComparatorClass() or MRJobConfig.GROUP_COMPARATOR_CLASS |
| // TODO: Check why tez-mapreduce ReduceProcessor use two different tez |
| // settings for the same MRJobConfig.GROUP_COMPARATOR_CLASS and use only one |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS, |
| comparatorClass); |
| conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS, |
| comparatorClass); |
| } |
| |
| private void setOutputFormat(org.apache.hadoop.mapreduce.Job job) { |
| // the OutputFormat we report to Hadoop is always PigOutputFormat which |
| // can be wrapped with LazyOutputFormat provided if PigConfiguration.PIG_OUTPUT_LAZY is set |
| if ("true".equalsIgnoreCase(job.getConfiguration().get(PigConfiguration.PIG_OUTPUT_LAZY))) { |
| LazyOutputFormat.setOutputFormatClass(job,PigOutputFormatTez.class); |
| } else { |
| job.setOutputFormatClass(PigOutputFormatTez.class); |
| } |
| } |
| |
| } |
