| /* |
| * 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.mapReduceLayer; |
| |
| import java.io.ByteArrayOutputStream; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Properties; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.io.Text; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.mapreduce.InputSplit; |
| import org.apache.hadoop.mapreduce.Mapper; |
| import org.apache.log4j.PropertyConfigurator; |
| import org.apache.pig.PigConstants; |
| import org.apache.pig.PigException; |
| import org.apache.pig.backend.executionengine.ExecException; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result; |
| 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.physicalLayer.util.PlanHelper; |
| import org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil; |
| import org.apache.pig.data.DataBag; |
| import org.apache.pig.data.SchemaTupleBackend; |
| import org.apache.pig.data.Tuple; |
| import org.apache.pig.data.TupleFactory; |
| import org.apache.pig.impl.PigContext; |
| import org.apache.pig.impl.PigImplConstants; |
| import org.apache.pig.impl.io.PigNullableWritable; |
| 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.SpillableMemoryManager; |
| import org.apache.pig.impl.util.Utils; |
| import org.apache.pig.tools.pigstats.PigStatusReporter; |
| |
| /** |
| * This class is the base class for PigMapBase, which has slightly |
| * difference among different versions of hadoop. PigMapBase implementation |
| * is located in $PIG_HOME/shims. |
| **/ |
| public abstract class PigGenericMapBase extends Mapper<Text, Tuple, PigNullableWritable, Writable> { |
| |
| private final Log log = LogFactory.getLog(getClass()); |
| |
| protected byte keyType; |
| |
| //Map Plan |
| protected PhysicalPlan mp = null; |
| |
| // Store operators |
| protected List<POStore> stores; |
| |
| protected TupleFactory tf = TupleFactory.getInstance(); |
| |
| boolean inIllustrator = false; |
| |
| Context outputCollector; |
| |
| // Reporter that will be used by operators |
| // to transmit heartbeat |
| ProgressableReporter pigReporter; |
| |
| protected boolean errorInMap = false; |
| |
| PhysicalOperator[] roots; |
| |
| private PhysicalOperator leaf; |
| |
| private volatile boolean initialized = false; |
| |
| /** |
| * for local map/reduce simulation |
| * @param plan the map plan |
| */ |
| public void setMapPlan(PhysicalPlan plan) { |
| mp = plan; |
| } |
| |
| /** |
| * Will be called when all the tuples in the input |
| * are done. So reporter thread should be closed. |
| */ |
| @Override |
| public void cleanup(Context context) throws IOException, InterruptedException { |
| super.cleanup(context); |
| if(errorInMap) { |
| //error in map - returning |
| return; |
| } |
| |
| if(PigMapReduce.sJobConfInternal.get().get(JobControlCompiler.END_OF_INP_IN_MAP, "false").equals("true") && !mp.isEmpty()) { |
| // If there is a stream in the pipeline or if this map job belongs to merge-join we could |
| // potentially have more to process - so lets |
| // set the flag stating that all map input has been sent |
| // already and then lets run the pipeline one more time |
| // This will result in nothing happening in the case |
| // where there is no stream or it is not a merge-join in the pipeline |
| mp.endOfAllInput = true; |
| runPipeline(leaf); |
| } |
| |
| if (!inIllustrator) { |
| for (POStore store: stores) { |
| if (!initialized) { |
| MapReducePOStoreImpl impl |
| = new MapReducePOStoreImpl(context); |
| store.setStoreImpl(impl); |
| store.setUp(); |
| } |
| store.tearDown(); |
| } |
| } |
| |
| //Calling EvalFunc.finish() |
| UDFFinishVisitor finisher = new UDFFinishVisitor(mp, new DependencyOrderWalker<PhysicalOperator, PhysicalPlan>(mp)); |
| try { |
| finisher.visit(); |
| } catch (VisitorException e) { |
| int errCode = 2121; |
| String msg = "Error while calling finish method on UDFs."; |
| throw new VisitorException(msg, errCode, PigException.BUG, e); |
| } |
| |
| mp = null; |
| |
| PhysicalOperator.setReporter(null); |
| initialized = false; |
| } |
| |
| /** |
| * Configures the mapper with the map plan and the |
| * reproter thread |
| */ |
| @SuppressWarnings("unchecked") |
| @Override |
| public void setup(Context context) throws IOException, InterruptedException { |
| super.setup(context); |
| |
| Configuration job = context.getConfiguration(); |
| SpillableMemoryManager.getInstance().configure(job); |
| context.getConfiguration().set(PigConstants.TASK_INDEX, Integer.toString(context.getTaskAttemptID().getTaskID().getId())); |
| PigMapReduce.sJobContext = context; |
| PigMapReduce.sJobConfInternal.set(context.getConfiguration()); |
| PigMapReduce.sJobConf = context.getConfiguration(); |
| inIllustrator = inIllustrator(context); |
| |
| PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(job.get("udf.import.list"))); |
| |
| // This attempts to fetch all of the generated code from the distributed cache, and resolve it |
| SchemaTupleBackend.initialize(job); |
| |
| Properties log4jProperties = (Properties) ObjectSerializer |
| .deserialize(job.get(PigImplConstants.PIG_LOG4J_PROPERTIES)); |
| if (log4jProperties != null) { |
| PropertyConfigurator.configure(log4jProperties); |
| } |
| |
| if (mp == null) |
| mp = (PhysicalPlan) ObjectSerializer.deserialize( |
| job.get("pig.mapPlan")); |
| stores = PlanHelper.getPhysicalOperators(mp, POStore.class); |
| |
| // To be removed |
| if(mp.isEmpty()) |
| log.debug("Map Plan empty!"); |
| else{ |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| mp.explain(baos); |
| log.debug(baos.toString()); |
| } |
| keyType = ((byte[])ObjectSerializer.deserialize(job.get("pig.map.keytype")))[0]; |
| // till here |
| |
| pigReporter = new ProgressableReporter(); |
| // Get the UDF specific context |
| MapRedUtil.setupUDFContext(job); |
| |
| if(!(mp.isEmpty())) { |
| |
| PigSplit split = (PigSplit)context.getInputSplit(); |
| List<OperatorKey> targetOpKeys = split.getTargetOps(); |
| |
| ArrayList<PhysicalOperator> targetOpsAsList = new ArrayList<PhysicalOperator>(); |
| for (OperatorKey targetKey : targetOpKeys) { |
| targetOpsAsList.add(mp.getOperator(targetKey)); |
| } |
| roots = targetOpsAsList.toArray(new PhysicalOperator[1]); |
| leaf = mp.getLeaves().get(0); |
| } |
| |
| PigStatusReporter pigStatusReporter = PigStatusReporter.getInstance(); |
| pigStatusReporter.setContext(new MRTaskContext(context)); |
| |
| log.info("Aliases being processed per job phase (AliasName[line,offset]): " + job.get("pig.alias.location")); |
| |
| Utils.setDefaultTimeZone(PigMapReduce.sJobConfInternal.get()); |
| } |
| |
| /** |
| * The map function that attaches the inpTuple appropriately |
| * and executes the map plan if its not empty. Collects the |
| * result of execution into oc or the input directly to oc |
| * if map plan empty. The collection is left abstract for the |
| * map-only or map-reduce job to implement. Map-only collects |
| * the tuple as-is whereas map-reduce collects it after extracting |
| * the key and indexed tuple. |
| */ |
| @Override |
| protected void map(Text key, Tuple inpTuple, Context context) throws IOException, InterruptedException { |
| if(!initialized) { |
| initialized = true; |
| // cache the collector for use in runPipeline() which |
| // can be called from close() |
| this.outputCollector = context; |
| pigReporter.setRep(context); |
| PhysicalOperator.setReporter(pigReporter); |
| |
| boolean aggregateWarning = "true".equalsIgnoreCase(context.getConfiguration().get("aggregate.warning")); |
| PigStatusReporter pigStatusReporter = PigStatusReporter.getInstance(); |
| pigStatusReporter.setContext(new MRTaskContext(context)); |
| PigHadoopLogger pigHadoopLogger = PigHadoopLogger.getInstance(); |
| pigHadoopLogger.setReporter(pigStatusReporter); |
| pigHadoopLogger.setAggregate(aggregateWarning); |
| PhysicalOperator.setPigLogger(pigHadoopLogger); |
| |
| if (!inIllustrator) { |
| for (POStore store: stores) { |
| MapReducePOStoreImpl impl |
| = new MapReducePOStoreImpl(context); |
| store.setStoreImpl(impl); |
| store.setUp(); |
| } |
| } |
| } |
| |
| if (mp.isEmpty()) { |
| collect(context,inpTuple); |
| return; |
| } |
| |
| for (PhysicalOperator root : roots) { |
| if (inIllustrator) { |
| if (root != null) { |
| root.attachInput(inpTuple); |
| } |
| } else { |
| root.attachInput(tf.newTupleNoCopy(inpTuple.getAll())); |
| } |
| } |
| |
| runPipeline(leaf); |
| } |
| |
| protected void runPipeline(PhysicalOperator leaf) throws IOException, InterruptedException { |
| while(true){ |
| Result res = leaf.getNextTuple(); |
| if(res.returnStatus==POStatus.STATUS_OK){ |
| collect(outputCollector,(Tuple)res.result); |
| continue; |
| } |
| |
| if(res.returnStatus==POStatus.STATUS_EOP) { |
| return; |
| } |
| |
| if(res.returnStatus==POStatus.STATUS_NULL) |
| continue; |
| |
| if(res.returnStatus==POStatus.STATUS_ERR){ |
| // remember that we had an issue so that in |
| // close() we can do the right thing |
| errorInMap = true; |
| // if there is an errmessage use it |
| String errMsg; |
| if(res.result != null) { |
| errMsg = "Received Error while " + |
| "processing the map plan: " + res.result; |
| } else { |
| errMsg = "Received Error while " + |
| "processing the map plan."; |
| } |
| |
| int errCode = 2055; |
| ExecException ee = new ExecException(errMsg, errCode, PigException.BUG); |
| throw ee; |
| } |
| } |
| |
| } |
| |
| abstract public void collect(Context oc, Tuple tuple) throws InterruptedException, IOException; |
| |
| abstract public boolean inIllustrator(Context context); |
| |
| /** |
| * @return the keyType |
| */ |
| public byte getKeyType() { |
| return keyType; |
| } |
| |
| /** |
| * @param keyType the keyType to set |
| */ |
| public void setKeyType(byte keyType) { |
| this.keyType = keyType; |
| } |
| |
| abstract public Context getIllustratorContext(Configuration conf, DataBag input, |
| List<Pair<PigNullableWritable, Writable>> output, InputSplit split) |
| throws IOException, InterruptedException; |
| } |
