text:
package org.apache.pig.backend.hadoop.executionengine.tez;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
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.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.v2.util.MRApps;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.pig.LoadFunc;
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.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.InputSizeReducerEstimator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingBigDecimalWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingBigIntegerWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingBooleanWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingCharArrayWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingDBAWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingDateTimeWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingDoubleWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingFloatWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingIntWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingLongWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingPartitionWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigGroupingTupleWritableComparator;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobControlCompiler.PigSecondaryKeyGroupComparator;
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.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.TezPOPackageAnnotator.LoRearrangeDiscoverer;
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.util.ParallelConstantVisitor;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.FileSpec;
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.NodeIdGenerator;
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.UDFContext;
import org.apache.tez.common.TezJobConfig;
import org.apache.tez.common.TezUtils;
import org.apache.tez.dag.api.DAG;
import org.apache.tez.dag.api.Edge;
import org.apache.tez.dag.api.EdgeManagerDescriptor;
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.OutputDescriptor;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.Vertex;
import org.apache.tez.dag.api.VertexGroup;
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.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.runtime.library.input.ConcatenatedMergedKeyValueInput;
import org.apache.tez.runtime.library.input.ShuffledMergedInput;
import org.apache.tez.runtime.library.input.SortedGroupedMergedInput;

/**
 * A visitor to construct DAG out of Tez plan.
 */
public class TezDagBuilder extends TezOpPlanVisitor {
    private static final Log log = LogFactory.getLog(TezJobControlCompiler.class);

    private static final String REDUCER_ESTIMATOR_KEY = "pig.exec.reducer.estimator";
    private static final String REDUCER_ESTIMATOR_ARG_KEY =  "pig.exec.reducer.estimator.arg";

    private DAG dag;
    private Map<String, LocalResource> localResources;
    private PigContext pc;
    private Configuration globalConf;

    private String scope;
    private NodeIdGenerator nig;

    public TezDagBuilder(PigContext pc, TezOperPlan plan, DAG dag,
            Map<String, LocalResource> localResources) {
        super(plan, new DependencyOrderWalker<TezOperator, TezOperPlan>(plan));
        this.pc = pc;
        this.globalConf = ConfigurationUtil.toConfiguration(pc.getProperties(), true);
        this.localResources = localResources;
        this.dag = dag;
        this.scope = plan.getRoots().get(0).getOperatorKey().getScope();
        this.nig = NodeIdGenerator.getGenerator();

        try {
            // Add credentials from binary token file and get tokens for namenodes
            // specified in mapreduce.job.hdfs-servers
            SecurityHelper.populateTokenCache(globalConf, dag.getCredentials());
        } catch (IOException e) {
            throw new RuntimeException("Error while fetching delegation tokens", e);
        }
    }

    @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()) {
                boolean isMap = (predecessors == null || predecessors.isEmpty()) ? true : false;
                to = newVertex(tezOp, isMap);
                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);
                            Edge edge = new Edge(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) {
                    vertexGroup.addOutput(store.getOperatorKey().toString(),
                            tezOp.getVertexGroupInfo().getStoreOutputDescriptor(),
                            MROutputCommitter.class);
                }
            }
        }
    }

    private GroupInputEdge newGroupInputEdge(TezOperator fromOp,
            TezOperator toOp, VertexGroup from, Vertex to) throws IOException {

        EdgeProperty edgeProperty = newEdge(fromOp, toOp);

        String groupInputClass = ConcatenatedMergedKeyValueInput.class.getName();

        // In case of SCATTER_GATHER and ShuffledUnorderedKVInput it will still be
        // ConcatenatedMergedKeyValueInput
        if(edgeProperty.getDataMovementType().equals(DataMovementType.SCATTER_GATHER)
                && edgeProperty.getEdgeDestination().getClassName().equals(ShuffledMergedInput.class.getName())) {
            groupInputClass = SortedGroupedMergedInput.class.getName();
        }

        return new GroupInputEdge(from, to, edgeProperty,
                new InputDescriptor(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)
            throws IOException {
        TezEdgeDescriptor edge = to.inEdges.get(from.getOperatorKey());
        PhysicalPlan combinePlan = edge.combinePlan;

        InputDescriptor in = new InputDescriptor(edge.inputClassName);
        OutputDescriptor out = new OutputDescriptor(edge.outputClassName);

        Configuration conf = ConfigurationUtil.toConfiguration(pc.getProperties(), false);
        if (!combinePlan.isEmpty()) {
            addCombiner(combinePlan, to, conf);
        }

        List<POLocalRearrangeTez> lrs = PlanHelper.getPhysicalOperators(from.plan,
                POLocalRearrangeTez.class);

        for (POLocalRearrangeTez lr : lrs) {
            if (lr.getOutputKey().equals(to.getOperatorKey().toString())) {
                byte keyType = lr.getKeyType();
                setIntermediateOutputKeyValue(keyType, conf, to, lr.isConnectedToPackage());
                // 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(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS,
                MRPartitioner.class.getName());

        if (edge.getIntermediateOutputKeyClass() != null) {
            conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
                    edge.getIntermediateOutputKeyClass());
        }

        if (edge.getIntermediateOutputValueClass() != null) {
            conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS,
                    edge.getIntermediateOutputValueClass());
        }

        if (edge.getIntermediateOutputKeyComparatorClass() != null) {
            conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
                    edge.getIntermediateOutputKeyComparatorClass());
        }

        conf.setBoolean("mapred.mapper.new-api", true);
        conf.set("pig.pigContext", ObjectSerializer.serialize(pc));
        conf.set("udf.import.list",
                ObjectSerializer.serialize(PigContext.getPackageImportList()));

        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(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
                    PigSecondaryKeyComparator.class.getName());
            // In MR - job.setOutputKeyClass() or MRJobConfig.OUTPUT_KEY_CLASS
            conf.set(TezJobConfig.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());
        }

        conf.set("udf.import.list",
                ObjectSerializer.serialize(PigContext.getPackageImportList()));

        MRToTezHelper.processMRSettings(conf, globalConf);

        in.setUserPayload(TezUtils.createUserPayloadFromConf(conf));
        out.setUserPayload(TezUtils.createUserPayloadFromConf(conf));

        if (to.getEstimatedParallelism()!=-1 && (to.isGlobalSort()||to.isSkewedJoin())) {
            // Use custom edge
            return new EdgeProperty((EdgeManagerDescriptor)null,
                    edge.dataSourceType, edge.schedulingType, out, in);
            }

        return new EdgeProperty(edge.dataMovementType, edge.dataSourceType,
                edge.schedulingType, out, in);
    }

    private void addCombiner(PhysicalPlan combinePlan, TezOperator pkgTezOp,
            Configuration conf) throws IOException {
        POPackage combPack = (POPackage) combinePlan.getRoots().get(0);
        POLocalRearrange combRearrange = (POLocalRearrange) combinePlan
                .getLeaves().get(0);
        setIntermediateOutputKeyValue(combRearrange.getKeyType(), conf, pkgTezOp);

        LoRearrangeDiscoverer lrDiscoverer = new LoRearrangeDiscoverer(
                combinePlan, pkgTezOp, combPack);
        lrDiscoverer.visit();

        combinePlan.remove(combPack);
        conf.set(TezJobConfig.TEZ_RUNTIME_COMBINER_CLASS,
                MRCombiner.class.getName());
        conf.set(MRJobConfig.COMBINE_CLASS_ATTR,
                PigCombiner.Combine.class.getName());
        conf.setBoolean("mapred.mapper.new-api", true);
        conf.set("pig.pigContext", ObjectSerializer.serialize(pc));
        conf.set("udf.import.list",
                ObjectSerializer.serialize(PigContext.getPackageImportList()));
        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, boolean isMap) throws IOException,
            ClassNotFoundException, InterruptedException {
        ProcessorDescriptor procDesc = new ProcessorDescriptor(
                tezOp.getProcessorName());

        // Pass physical plans to vertex as user payload.
        JobConf payloadConf = new JobConf(ConfigurationUtil.toConfiguration(pc.getProperties(), false));

        // 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();

        if (tezOp.sampleOperator != null) {
            payloadConf.set(PigProcessor.SAMPLE_VERTEX, tezOp.sampleOperator.getOperatorKey().toString());
        }

        if (tezOp.sortOperator != null) {
            payloadConf.set(PigProcessor.SORT_VERTEX, tezOp.sortOperator.getOperatorKey().toString());
        }

        String tmp;
        long maxCombinedSplitSize = 0;
        if (!tezOp.combineSmallSplits() || pc.getProperties().getProperty(PigConfiguration.PIG_SPLIT_COMBINATION, "true").equals("false"))
            payloadConf.setBoolean(PigConfiguration.PIG_NO_SPLIT_COMBINATION, true);
        else if ((tmp = pc.getProperties().getProperty(PigConfiguration.PIG_MAX_COMBINED_SPLIT_SIZE, null)) != null) {
            try {
                maxCombinedSplitSize = Long.parseLong(tmp);
            } catch (NumberFormatException e) {
                log.warn("Invalid numeric format for pig.maxCombinedSplitSize; use the default maximum combined split size");
            }
        }
        if (maxCombinedSplitSize > 0)
            payloadConf.setLong("pig.maxCombinedSplitSize", maxCombinedSplitSize);

        List<POLoad> loads = processLoads(tezOp, payloadConf, job);
        LinkedList<POStore> stores = processStores(tezOp, payloadConf, job);

        payloadConf.set("pig.pigContext", ObjectSerializer.serialize(pc));
        payloadConf.set("udf.import.list",
                ObjectSerializer.serialize(PigContext.getPackageImportList()));
        payloadConf.set("exectype", "TEZ");
        payloadConf.setBoolean("mapred.mapper.new-api", true);
        payloadConf.setClass("mapreduce.inputformat.class",
                PigInputFormat.class, InputFormat.class);

        // 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));
            setIntermediateOutputKeyValue(keyType, payloadConf, tezOp);
            POShuffleTezLoad newPack;
            newPack = new POShuffleTezLoad(pack);
            if (tezOp.isSkewedJoin()) {
                newPack.setSkewedJoins(true);
            }
            tezOp.plan.add(newPack);

            // 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>();
            for (TezOperator pred : mPlan.getPredecessors(tezOp)) {
                if (tezOp.sampleOperator != null && tezOp.sampleOperator == pred) {
                    // skip sample vertex input
                } else {
                    String inputKey = pred.getOperatorKey().toString();
                    if (pred.isVertexGroup()) {
                        pred = mPlan.getOperator(pred.getVertexGroupMembers().get(0));
                    }
                    LinkedList<POLocalRearrangeTez> lrs =
                            PlanHelper.getPhysicalOperators(pred.plan, POLocalRearrangeTez.class);
                    for (POLocalRearrangeTez lr : lrs) {
                        if (lr.isConnectedToPackage()
                                && lr.getOutputKey().equals(tezOp.getOperatorKey().toString())) {
                            localRearrangeMap.put((int) lr.getIndex(), inputKey);
                        }
                    }
                }
            }
            for (Map.Entry<Integer, String> entry : localRearrangeMap.entrySet()) {
                newPack.addInputKey(entry.getValue());
            }

            if (succsList != null) {
                for (PhysicalOperator succs : succsList) {
                    tezOp.plan.connect(newPack, succs);
                }
            }

            setIntermediateOutputKeyValue(pack.getPkgr().getKeyType(), payloadConf, tezOp);
        } else if (roots.size() == 1 && roots.get(0) instanceof POIdentityInOutTez) {
            POIdentityInOutTez identityInOut = (POIdentityInOutTez) roots.get(0);
            // TODO Need to fix multiple input key mapping
            TezOperator identityInOutPred = null;
            for (TezOperator pred : mPlan.getPredecessors(tezOp)) {
                if (!pred.isSampleAggregation()) {
                    identityInOutPred = pred;
                    break;
                }
            }
            identityInOut.setInputKey(identityInOutPred.getOperatorKey().toString());
        } else if (roots.size() == 1 && roots.get(0) instanceof POValueInputTez) {
            POValueInputTez valueInput = (POValueInputTez) roots.get(0);

            LinkedList<String> scalarInputs = new LinkedList<String>();
            for (POUserFunc userFunc : PlanHelper.getPhysicalOperators(tezOp.plan, POUserFunc.class) ) {
                if (userFunc.getFunc() instanceof ReadScalarsTez) {
                    scalarInputs.add(((ReadScalarsTez)userFunc.getFunc()).getTezInputs()[0]);
                }
            }
            // Make sure we don't find the scalar
            for (TezOperator pred : mPlan.getPredecessors(tezOp)) {
                if (!scalarInputs.contains(pred.getOperatorKey().toString())) {
                    valueInput.setInputKey(pred.getOperatorKey().toString());
                    break;
                }
            }
        }
        JobControlCompiler.setOutputFormat(job);

        // 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));

        UDFContext.getUDFContext().serialize(payloadConf);

        MRToTezHelper.processMRSettings(payloadConf, globalConf);

        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));
        }

        // Can only set parallelism here if the parallelism isn't derived from
        // splits
        int parallelism = -1;
        InputSplitInfo inputSplitInfo = null;
        if (loads != null && loads.size() > 0) {
            // Not using MRInputAMSplitGenerator because delegation tokens are
            // fetched in FileInputFormat
            inputSplitInfo = MRHelpers.generateInputSplitsToMem(payloadConf);
            // TODO: Can be set to -1 if TEZ-601 gets fixed and getting input
            // splits can be moved to if(loads) block below
            parallelism = inputSplitInfo.getNumTasks();
            tezOp.setRequestedParallelism(parallelism);
        } else {
            int prevParallelism = -1;
            boolean isOneToOneParallelism = false;
            for (Map.Entry<OperatorKey,TezEdgeDescriptor> entry : tezOp.inEdges.entrySet()) {
                if (entry.getValue().dataMovementType == DataMovementType.ONE_TO_ONE) {
                    TezOperator pred = mPlan.getOperator(entry.getKey());
                    parallelism = pred.getEffectiveParallelism();
                    if (prevParallelism == -1) {
                        prevParallelism = parallelism;
                    } else if (prevParallelism != parallelism) {
                        throw new IOException("one to one sources parallelism for vertex "
                                + tezOp.getOperatorKey().toString() + " are not equal");
                    }
                    if (pred.getRequestedParallelism()!=-1) {
                        tezOp.setRequestedParallelism(pred.getRequestedParallelism());
                    } else {
                        tezOp.setEstimatedParallelism(pred.getEstimatedParallelism());
                    }
                    isOneToOneParallelism = true;
                    parallelism = -1;
                }
            }
            if (!isOneToOneParallelism) {
                if (tezOp.getRequestedParallelism()!=-1) {
                    parallelism = tezOp.getRequestedParallelism();
                } else if (pc.defaultParallel!=-1) {
                    parallelism = pc.defaultParallel;
                } else {
                    parallelism = estimateParallelism(job, mPlan, tezOp);
                    tezOp.setEstimatedParallelism(parallelism);
                    if (tezOp.isGlobalSort()||tezOp.isSkewedJoin()) {
                        // Vertex manager will set parallelism
                        parallelism = -1;
                    }
                }
            }
        }

        // Once we decide the parallelism of the sampler, propagate to
        // downstream operators if necessary
        if (tezOp.isSampler()) {
            // There could be multiple sampler and share the same sample aggregation job
            // and partitioner job
            TezOperator sampleAggregationOper = null;
            TezOperator sampleBasedPartionerOper = null;
            TezOperator sortOper = null;
            for (TezOperator succ : mPlan.getSuccessors(tezOp)) {
                if (succ.isVertexGroup()) {
                    succ = mPlan.getSuccessors(succ).get(0);
                }
                if (succ.isSampleAggregation()) {
                    sampleAggregationOper = succ;
                } else if (succ.isSampleBasedPartitioner()) {
                    sampleBasedPartionerOper = succ;
                }
            }
            sortOper = mPlan.getSuccessors(sampleBasedPartionerOper).get(0);

            if (sortOper.getRequestedParallelism()==-1 && pc.defaultParallel==-1) {
                // set estimate parallelism for order by/skewed join to sampler parallelism
                // that include:
                // 1. sort operator
                // 2. constant for sample aggregation oper
                sortOper.setEstimatedParallelism(parallelism);
                ParallelConstantVisitor visitor =
                        new ParallelConstantVisitor(sampleAggregationOper.plan, parallelism);
                visitor.visit();
            }
        }

        if (tezOp.isNeedEstimateParallelism()) {
            payloadConf.setBoolean(PigProcessor.ESTIMATE_PARALLELISM, true);
            log.info("Estimate quantile for sample aggregation vertex " + tezOp.getOperatorKey().toString());
        }

        // Take our assembled configuration and create a vertex
        byte[] userPayload = TezUtils.createUserPayloadFromConf(payloadConf);
        procDesc.setUserPayload(userPayload);

        Vertex vertex = new Vertex(tezOp.getOperatorKey().toString(), procDesc, parallelism,
                isMap ? MRHelpers.getMapResource(globalConf) : MRHelpers.getReduceResource(globalConf));

        Map<String, String> taskEnv = new HashMap<String, String>();
        MRHelpers.updateEnvironmentForMRTasks(globalConf, taskEnv, isMap);
        vertex.setTaskEnvironment(taskEnv);

        // 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,
                job.getCredentials());
        MRApps.setupDistributedCache(globalConf, localResources);
        vertex.setTaskLocalFiles(localResources);

        vertex.setTaskLaunchCmdOpts(isMap ? MRHelpers.getMapJavaOpts(globalConf)
                : MRHelpers.getReduceJavaOpts(globalConf));

        log.info("For vertex - " + tezOp.getOperatorKey().toString()
                + ": parallelism=" + parallelism
                + ", memory=" + vertex.getTaskResource().getMemory()
                + ", java opts=" + vertex.getTaskLaunchCmdOpts()
                );

        // 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 : loads) {

            // TODO: These should get the globalConf, or a merged version that
            // keeps settings like pig.maxCombinedSplitSize
            vertex.setTaskLocationsHint(inputSplitInfo.getTaskLocationHints());
            vertex.addInput(ld.getOperatorKey().toString(),
                    new InputDescriptor(MRInput.class.getName())
                            .setUserPayload(MRHelpers.createMRInputPayload(
                                    userPayload,
                                    inputSplitInfo.getSplitsProto())),
                    MRInputSplitDistributor.class);
        }

        for (POStore store : stores) {

            ArrayList<POStore> emptyList = new ArrayList<POStore>();
            ArrayList<POStore> singleStore = new ArrayList<POStore>();
            singleStore.add(store);

            Configuration outputPayLoad = new Configuration(payloadConf);
            outputPayLoad.set(JobControlCompiler.PIG_MAP_STORES,
                    ObjectSerializer.serialize(emptyList));
            outputPayLoad.set(JobControlCompiler.PIG_REDUCE_STORES,
                    ObjectSerializer.serialize(singleStore));

            OutputDescriptor storeOutDescriptor = new OutputDescriptor(
                    MROutput.class.getName()).setUserPayload(TezUtils
                    .createUserPayloadFromConf(outputPayLoad));
            if (tezOp.getVertexGroupStores() != null) {
                OperatorKey vertexGroupKey = tezOp.getVertexGroupStores().get(store.getOperatorKey());
                if (vertexGroupKey != null) {
                    getPlan().getOperator(vertexGroupKey).getVertexGroupInfo()
                            .setStoreOutputDescriptor(storeOutDescriptor);
                    continue;
                }
            }
            vertex.addOutput(store.getOperatorKey().toString(),
                    storeOutDescriptor, MROutputCommitter.class);
        }

        // 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);
        }

        // Set the right VertexManagerPlugin
        if (tezOp.getEstimatedParallelism() != -1) {
            if (tezOp.isGlobalSort()||tezOp.isSkewedJoin()) {
                // Set VertexManagerPlugin to PartitionerDefinedVertexManager, which is able
                // to decrease/increase parallelism of sorting vertex dynamically
                // based on the numQuantiles calculated by sample aggregation vertex
                vertex.setVertexManagerPlugin(new VertexManagerPluginDescriptor(
                        PartitionerDefinedVertexManager.class.getName()));
                log.info("Set VertexManagerPlugin to PartitionerDefinedParallelismVertexManager for vertex " + tezOp.getOperatorKey().toString());
            } else {
                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 && !containCustomPartitioner) {
                    // Use auto-parallelism feature of ShuffleVertexManager to dynamically
                    // reduce the parallelism of the vertex
                    VertexManagerPluginDescriptor vmPluginDescriptor = new VertexManagerPluginDescriptor(
                            ShuffleVertexManager.class.getName());
                    Configuration vmPluginConf = ConfigurationUtil.toConfiguration(pc.getProperties(), false);
                    vmPluginConf.setBoolean(ShuffleVertexManager.TEZ_AM_SHUFFLE_VERTEX_MANAGER_ENABLE_AUTO_PARALLEL, true);
                    if (vmPluginConf.getLong(InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM,
                            InputSizeReducerEstimator.DEFAULT_BYTES_PER_REDUCER)!=
                                    InputSizeReducerEstimator.DEFAULT_BYTES_PER_REDUCER) {
                        vmPluginConf.setLong(ShuffleVertexManager.TEZ_AM_SHUFFLE_VERTEX_MANAGER_DESIRED_TASK_INPUT_SIZE,
                                vmPluginConf.getLong(InputSizeReducerEstimator.BYTES_PER_REDUCER_PARAM,
                                        InputSizeReducerEstimator.DEFAULT_BYTES_PER_REDUCER));
                    }
                    vmPluginDescriptor.setUserPayload(TezUtils.createUserPayloadFromConf(vmPluginConf));
                    vertex.setVertexManagerPlugin(vmPluginDescriptor);
                    log.info("Set auto parallelism for vertex " + tezOp.getOperatorKey().toString());
                }
            }
        }

        // Reset udfcontext jobconf. It is not supposed to be set in the front end
        UDFContext.getUDFContext().addJobConf(null);
        return vertex;
    }

    /**
     * Do the final configuration of LoadFuncs and store what goes where. This
     * will need to be changed as the inputs get un-bundled
     *
     * @param tezOp
     * @param conf
     * @param job
     * @return true if any POLoads were found, else false.
     * @throws VisitorException
     * @throws IOException
     */
    private List<POLoad> processLoads(TezOperator tezOp, Configuration conf,
            Job job) throws VisitorException, IOException {
        ArrayList<FileSpec> inp = new ArrayList<FileSpec>();
        ArrayList<List<OperatorKey>> inpTargets = new ArrayList<List<OperatorKey>>();
        ArrayList<String> inpSignatureLists = new ArrayList<String>();
        ArrayList<Long> inpLimits = new ArrayList<Long>();

        List<POLoad> lds = PlanHelper.getPhysicalOperators(tezOp.plan,
                POLoad.class);

        if (lds != null && lds.size() > 0) {
            for (POLoad ld : lds) {
                LoadFunc lf = ld.getLoadFunc();
                lf.setLocation(ld.getLFile().getFileName(), job);

                // Store the inp filespecs
                inp.add(ld.getLFile());
            }
        }

        if (lds != null && lds.size() > 0) {
            for (POLoad ld : lds) {
                // Store the target operators for tuples read
                // from this input
                List<PhysicalOperator> ldSucs = new ArrayList<PhysicalOperator>(
                        tezOp.plan.getSuccessors(ld));
                List<OperatorKey> ldSucKeys = new ArrayList<OperatorKey>();
                if (ldSucs != null) {
                    for (PhysicalOperator operator2 : ldSucs) {
                        ldSucKeys.add(operator2.getOperatorKey());
                    }
                }
                inpTargets.add(ldSucKeys);
                inpSignatureLists.add(ld.getSignature());
                inpLimits.add(ld.getLimit());
                // Remove the POLoad from the plan
                tezOp.plan.remove(ld);
                // Now add the input handling operator for the Tez backend
                // TODO: Move this upstream to the PhysicalPlan generation
                POSimpleTezLoad tezLoad = new POSimpleTezLoad(new OperatorKey(
                        scope, nig.getNextNodeId(scope)), ld.getLFile());
                tezLoad.setInputKey(ld.getOperatorKey().toString());
                tezLoad.setAlias(ld.getAlias());
                tezOp.plan.add(tezLoad);
                for (PhysicalOperator sucs : ldSucs) {
                    tezOp.plan.connect(tezLoad, sucs);
                }

            }
        }

        conf.set("pig.inputs", ObjectSerializer.serialize(inp));
        conf.set("pig.inpTargets", ObjectSerializer.serialize(inpTargets));
        conf.set("pig.inpSignatures", ObjectSerializer.serialize(inpSignatureLists));
        conf.set("pig.inpLimits", ObjectSerializer.serialize(inpLimits));

        return lds;
    }

    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);
            }

            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;
    }

    private void setIntermediateOutputKeyValue(byte keyType, Configuration conf, TezOperator tezOp)
            throws JobCreationException, ExecException {
        setIntermediateOutputKeyValue(keyType, conf, tezOp, true);
    }

    @SuppressWarnings("rawtypes")
    private void setIntermediateOutputKeyValue(byte keyType, Configuration conf, TezOperator tezOp,
            boolean isConnectedToPackage) throws JobCreationException, ExecException {
        if (tezOp != null && tezOp.isUseSecondaryKey() && isConnectedToPackage) {
            conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
                    NullableTuple.class.getName());
        } else if (tezOp != null && tezOp.isSkewedJoin() && isConnectedToPackage) {
            conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
                    NullablePartitionWritable.class.getName());
        } else {
            Class<? extends WritableComparable> keyClass = HDataType
                    .getWritableComparableTypes(keyType).getClass();
            conf.set(TezJobConfig.TEZ_RUNTIME_KEY_CLASS,
                    keyClass.getName());

        }
        conf.set(TezJobConfig.TEZ_RUNTIME_VALUE_CLASS,
                NullableTuple.class.getName());
        conf.set(TezJobConfig.TEZ_RUNTIME_PARTITIONER_CLASS,
                MRPartitioner.class.getName());
        selectOutputComparator(keyType, conf, tezOp);
    }

    private static Class<? extends WritableComparator> comparatorForKeyType(byte keyType, boolean hasOrderBy)
            throws JobCreationException {

        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:
            //if (hasOrderBy) {
                return PigBytesRawComparator.class;
            //} else {
            //    return PigDBAWritableComparator.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:
            //TODO: PigTupleWritableComparator gives wrong results with cogroup in
            //Checkin_2 and few other e2e tests. But MR has PigTupleWritableComparator
            //Investigate the difference later
            //if (hasOrderBy) {
                return PigTupleSortComparator.class;
            //} else {
            //    return PigTupleWritableComparator.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> getGroupingComparatorForKeyType(byte keyType)
            throws JobCreationException {

        switch (keyType) {
        case DataType.BOOLEAN:
            return PigGroupingBooleanWritableComparator.class;

        case DataType.INTEGER:
            return PigGroupingIntWritableComparator.class;

        case DataType.BIGINTEGER:
            return PigGroupingBigIntegerWritableComparator.class;

        case DataType.BIGDECIMAL:
            return PigGroupingBigDecimalWritableComparator.class;

        case DataType.LONG:
            return PigGroupingLongWritableComparator.class;

        case DataType.FLOAT:
            return PigGroupingFloatWritableComparator.class;

        case DataType.DOUBLE:
            return PigGroupingDoubleWritableComparator.class;

        case DataType.DATETIME:
            return PigGroupingDateTimeWritableComparator.class;

        case DataType.CHARARRAY:
            return PigGroupingCharArrayWritableComparator.class;

        case DataType.BYTEARRAY:
            return PigGroupingDBAWritableComparator.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 PigGroupingTupleWritableComparator.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 selectOutputComparator(byte keyType, Configuration conf, TezOperator tezOp)
            throws JobCreationException {
        // TODO: Handle sorting like in JobControlCompiler
        // TODO: Group comparators as in JobControlCompiler
        if (tezOp != null && tezOp.isUseSecondaryKey()) {
            conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
                    PigSecondaryKeyComparator.class.getName());
            setGroupingComparator(conf, PigSecondaryKeyGroupComparator.class.getName());
        } else {
            if (tezOp != null && tezOp.isSkewedJoin()) {
                // TODO: PigGroupingPartitionWritableComparator only used as Group comparator in MR.
                // What should be TEZ_RUNTIME_KEY_COMPARATOR_CLASS if same as MR?
                conf.set(TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
                        PigGroupingPartitionWritableComparator.class.getName());
                setGroupingComparator(conf, PigGroupingPartitionWritableComparator.class.getName());
            } else {
                boolean hasOrderby = hasOrderby(tezOp);
                conf.setClass(
                        TezJobConfig.TEZ_RUNTIME_KEY_COMPARATOR_CLASS,
                        comparatorForKeyType(keyType, hasOrderby), RawComparator.class);
                if (!hasOrderby) {
                    setGroupingComparator(conf, getGroupingComparatorForKeyType(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(TezJobConfig.TEZ_RUNTIME_GROUP_COMPARATOR_CLASS,
                comparatorClass);
        conf.set(TezJobConfig.TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS,
                comparatorClass);
    }

    public static int estimateParallelism(Job job, TezOperPlan tezPlan,
            TezOperator tezOp) throws IOException {
        Configuration conf = job.getConfiguration();

        TezParallelismEstimator estimator = conf.get(REDUCER_ESTIMATOR_KEY) == null ? new TezOperDependencyParallelismEstimator()
                : PigContext.instantiateObjectFromParams(conf,
                        REDUCER_ESTIMATOR_KEY, REDUCER_ESTIMATOR_ARG_KEY,
                        TezParallelismEstimator.class);

        log.info("Using parallel estimator: " + estimator.getClass().getName());
        int numberOfReducers = estimator.estimateParallelism(tezPlan, tezOp, conf);
        return numberOfReducers;
    }

}

redpajama_set_name: RedPajamaGithub


Our Spiritual Growth Campaign for 2016 begins on January 10th. This year we'll be looking at Boundaries and the powerful effect they can have in your life.
For those of you who've never done a Spiritual Growth Campaign they are basically made up of three parts.
Part One is the Sermon on Sunday on the Campaign topic.
Part Two is a daily devotional on the Campaign topic.
Part Three is a weekly small group on the Campaign topic.
All this material has been done right here at CCR. Louie has written the devotionals & he and Amy have recorded the small group lessons for you to watch as well as prepared a workbook for you to use.
So get ready for this life changing event, and pray that God will use it to impact our community and to help us Know Him & Make Him Known.
All our video lessons will be embedded here in weekly updates once the Campaign starts. You can download the PDF of the Devotional & Workbook below. The sermons will be available on the Sermon Archive page.


Viewer : 4.220.050 by : Sony Music South.
Viewer : 17.038.779 by : Zee Music Company.
Viewer : 203.670.061 by : Tinie Tempah.
Viewer : 4.717.174 by : T-Series.
Viewer : 196.474 by : KaRrl123WTF.
Viewer : 150.175 by : KaRrl123WTF.
Viewer : 1.472.330 by : Sony Music South.
Viewer : 52.745 by : Country Fun Studio.
Viewer : 202.292 by : KaRrl123WTF.
Viewer : 133.667 by : KaRrl123WTF.
Viewer : 49.905.973 by : Lithuania HQ.
Viewer : 100.637 by : KaRrl123WTF.
Viewer : 50.574.846 by : NoCopyrightSounds.
Viewer : 8.438.552 by : Il Ribelle.
Viewer : 113.127 by : KaRrl123WTF.
Viewer : 37.976.212 by : Think Music India.
Viewer : 278.501 by : KaRrl123WTF.
Viewer : 125.130 by : KaRrl123WTF.
Viewer : 139.487 by : KaRrl123WTF.


Atlas Rug & Design Centre Inc.
Reviews for Atlas Rug & Design Centre Inc.
Our basement carpet was badly installed and old, smelly and musty. David came to the house, assessed the current situation, proposed an excellent solution, and did a great job removing the old carpet and backing; smoothing out the very uneven cement; and installing beautiful vinyl flooring that is attractive, durable and reasonably priced. He and Tony were polite, on time, within budget, cleaned up beautifully, and went above and beyond.
We highly recommend Atlas Rug for your flooring needs.
Both my home and rental property were installed by David. The hard floor was beautiful and top quality, and we got lots of good comments during our Open house sale (one of our house highlights). David knows his products very well and have good sense of matching existing furnitures and colors, he helped us choose the color and turned out better than we expected. The installer crews were also amazing, very on time and polite, both jobs were done professionally. The post sale customer services are excellent. After installation David called us to ensure we are happy about the floors and taught us how to maintain and care the flooring. Definitely I will use him again for any future flooring needs.
David was very helpful in assisting me with choosing appropriate flooring for my space and also with choosing rugs that would be suitable. He was very professional and always showed up on time when we had made appointments. The flooring was installed quickly and looks beautiful. When I was unable to find a rug for the bedroom that came in a standard size that suited the space, he was able to make one out of carpeting he had in the store. It was perfect. I used this company for floor and rugs on the main floor in 2007 as well and wouldn't consider even looking anywhere else. I recommended them to my 2 of my neighbours and they were also pleased with the floors and rugs they purchased.
EXCELLENT SERVICE AND QUALITY OF PRODUCT!
David, the owner, helped me select a 100% wool carpet to replace the 10 year old one in my condo bedroom. He also had an incredible high quality Oak wood flooring. His renovations alone have increased the value of my place by $15K and his quality of work was exceptional! Very competitive pricing and exceptional service.
We had fabulous service from Atlas...from selecting the carpet to the installation. When David came to do measurements for the stairs, he suggested a carpet that we had not seen. We had done our research and had 3 quotes. He showed us the end of a huge installation which was carpet outside of our budget but because it was an end, it was! It's gorgeous. The carpet was installed beautifully in an old home with wonky stairs. We are thrilled.
We didn't get a 2nd or 3rd estimate - which one should ALWAYS get. My husband liked the owner of Atlas Rug & Design Centre Inc., who was very charming. The quote given was $7683.00 (taxes included). We had no idea if that was reasonable or not, but it was within our budget. My husband was impatient and he thought in order not to waste time - he went ahead after just one estimate with Atlas Rug & Design Centre Inc.
However, after the deposit was paid and the rug was ordered, the salesman phoned and told us he MISSED calculating one entire room! That was another $1273.19 (taxes included) additional!
Workmanship was alright - that was another team.
Anyhow, my husband still gave this company another chance, when we wanted hardwood floors put in in another house. This time, I insisted on obtaining other estimates in order to compare - now, their estimate was some $3000+ more than another company which we eventually hired to do the job.
CARPET REPAIR IN COMMON ELEMENTS. CONTACT AND SERVICE WAS GOOD. MISCOMMUNICATIION ON COST. UNDERSTOOD CARPET WAS IN STOCK BUT IN THE END IT WAS NOT. STILL, THE WORK WAS GOOD. WOULD USE THEM AGAIN.
rgain...was it one...will tell with time on the wear and tear.
Convenient, extremely professional and good in terms of price.
We simply walked in to the store, saw the floor that we really really liked and got an installation for the next day. The installer was a hard worker and the whole thing got done in a day (~400 sq. ft. of an old and fairly crooked house - we are constantly told by contractors).
We love the floor, and everyone who comes by comments on how awesome it looks.
Atlas Rug & Design Centre Inc. has not added any photos.
More about Atlas Rug & Design Centre Inc.
Welcome to Atlas Rug & Design Centre Inc.


A bumper week for gaming is led by 'Horizon Zero Dawn' which tops the chart as Sony's biggest ever PS4 launch of a new IP (ahead of No Man's Sky).
Released on Wednesday 'Horizon Zero Dawn' is also the biggest launch of the year so far and Sony's biggest launch since 'Uncharted 4: A Thief's End'. Ubisoft's 'For Honor' (-45%) falls from No1 to No5, having to make way for both Horizon Zero Dawn and new Nintendo Switch titles. Released on Friday, Nintendo Switch brings with it 'The Legend of Zelda: The Breath of the Wild', which debuts at No2 for both Switch and Wii U. 78% of sales were for the Switch version with 22% for Wii U. The Switch version becomes the 3rd biggest ever launch of a Zelda title, behind 'Wind Waker' on Gamecube and 'Twilight Princess' on Wii. There were 5 new Switch games at launch, 3 of which enter the Top 10 including '1-2 Switch' new at No4 and Konami's 'Super Bomberman R' debuting at No7. The Switch version of 'Just Dance 2017' helps Ubisoft's game re-enter the Top 40 at No24 (+68%), while the Switch version of 'Skylanders Imaginators' fails to lift the title back in to the chart. The only other new game to make the Top 40 this week is Telltale's 'The Walking Dead: The New Frontier' which begins life at No19.


Soft skills are the most important skills you will need to achieve success in your professional and personal lives. So, what are soft skills?
Soft skills, which are likely the hardest skills of all, are the skills that characterize our relationships with ourselves and others. They are skills like self-awareness, collaboration, resilience, communication skills, and more. Even though 92% of executives believe soft skills are equally important or more important than technical skills, 89% of executives say that it is difficult to find people with these soft skills.
I have owned a soft skills training company called SkillsCamp for over 3 years now. Through SkillsCamp, I've had the pleasure of working with dozens of organizations from large Fortune 500s and start-ups to educational and governmental organizations at every level. Across every industry, we've helped people build these crucial soft skills and in doing so, I've gained insight into the most important soft skills across all industries. Though we offer a wide variety of skill competencies, these 5 are often the ones that come before the others. Whether you work alone, in a giant team, in a corporate office, or remotely, these skills matter.
If you work on developing these soft skills, I promise you will improve your performance and your well-being.
Self-awareness is often the golden skill underlying many soft skills. Before you can manage your interactions with others, you must first understand your own thoughts, beliefs, feelings, and actions. By understanding your strengths and weaknesses, how you cope with stress, or how you speak to others, you will be better able to tailor your actions to the people you're with. You will also be better able to understand your thresholds for stress thus making you better able to cope and bounce back in the face of difficulties.
To-Do: If you want to start today, try downloading and filling out some of SkillsCamp's self-awareness activities and explore your beliefs, values, motivators, and perception gaps. DOWNLOAD HERE.
Wouldn't it be nice if everyone magically had confidence? Unfortunately, that's not the case and I often see a lack of self-confidence become a barrier to learning other positive skills such as presentation skills or leadership. On the other side a lack of self-confidence can cause you to frame situations in ways that confirm your opinion of yourself (confirmation bias!). For example, you see two people whispering and believe they are making fun of you. In all reality, they could be talking about their kids or lunches. But, you may create a story in your head that causes you to shy away from those colleagues.
To-Do: This may be one of the hardest skills to build, but start seriously thinking about what makes you feel your best. Is it certain people? What is it about the way they treat you? Is it when you dress a certain way? Write these things down and try to add them to your life. Try positive affirmations and practice gratitude! Also, watch Dr. Amy Cuddy's TED talk and fake it until you become it!
After reviewing hundreds of successful people in all industries, we found that the number one quality they attributed to their success was being able to bounce back from failure and persevere in stress. How do you cope with stress? In the face of adversity or failure, how easy is it for you to overcome? When we do things that challenge us, stress is inevitable. We must learn to withstand stress because a lack of resilience can lead to dwelling on problems, feeling overwhelmed, more conflicts, using unhealthy coping tactics to handle stress, and developing anxiety and depression.
To-Do: There are many things you can do to build resilience, but for today, simply try a gratitude journal. Gratitude is one of the 5 factors of resilience and helps build optimism, another factor in resilience. Secondly, download and fill out one of SkillsCamp's resilience exercises around coping strategies. DOWNLOAD HERE.
Out of personal experience, a lack of resourcefulness or problem-solving capability has been the deficit that most annoys my clients. I hear things like "can't you just look it up on Google" or "can you stop bothering the manager with every little question?" The ability to navigate difficulties and find creative solutions to problems is key to success in any role at any level.
To-Do: GOOGLE IT. I'm not kidding here. It's actually incredibly surprising how many people don't use this tool – maybe they think it's not professional? IT IS. Can't figure out how to use the Photoshop template? Google it. Lost on your way to the team social? Google it. Do you have an internal resource hub? Use it!
Whether you're a manager, a team member, dealing with clients, or simply interact with people in general, emotional intelligence is increasingly becoming one of the most important skills any individual can have. While most workplace conflicts come from interpersonal conflict, emotional intelligence can help prevent these conflicts and promote a positive culture, higher job satisfaction, and more effective employees. Understanding another's feelings, needs, and motivators makes you a better team player, more adaptable, and a better salesperson too!
To-Do: Emotional intelligence can be broken down into self-awareness, self-regulation, social awareness, and relationship management, so for this action item, I will revert you back to the self-awareness activities. It must start here.
There a lot of soft skills we can build in ourselves, but these are definitely the top 5 that are important no matter your industry, job, or workplace culture. Start building them today! If you'd rather do it with your organization, reach out to SkillsCamp, and we'll make it happen!
SkillsCamp is a soft skills training company working with businesses and educational institutions to build soft skills in their staff and students — skills like resilience, collaboration, communication, and more. Visit us at skillscamp.co.
Bailey Parnell is the Founder & CEO of SkillsCamp, a soft skills training company, and was named one of Canada's Top 100 Most Powerful Women. Bailey is a TEDx speaker with near 1 million views, an award-winning digital marketer, and a businesswoman with a talent for helping people develop the skills they need for success. Her work and expertise have been featured in Forbes, CBC, FOX, Flare Magazine, and more. Bailey is also currently completing her part-time Masters in Communications and Culture at Ryerson University with research looking into social media's impact on mental health, the results of which she presented at the World Youth Forum in Egypt.


The entrance for Opening Doors Coaching is in the back of the building with the free parking lot. There is a sign beside the door and you will want to come up the stairs. The office is on the left at the top of the stairs.
Viktoria Munroe's office entrance is in the back of this building at 114 Commonwealth Avenue in West Concord. There is free parking in the lot behind the building where the entrance is.


text:
Here's why Emily dresses *like that* in 'Emily in Paris'
There's a reason why she wears bold colors and prints.
The show's costumes have a certain je ne sais quoi.Marie Etchegoyen / Netflix
Dec. 21, 2022, 3:42 PM UTC
By Chrissy Callahan
From juicy love triangles and office intrigue to scenic views and hot eye candy, there are beaucoup reasons why Netflix viewers are hooked on "Emily in Paris."
Bien sûr, the show's stunning fashion moments are another reason why fans keep coming back for more and they're in for a real sartorial treat as Season Three hits the streaming service this week.
To celebrate the show's return, TODAY.com caught up with "Emily in Paris" costume designer Marylin Fitoussi to break down this season's hottest fashion moments and take a look back at Emily's style evolution.
Fitoussi is especially proud of Season 3. / Netflix
On her style inspiration for Season 3
Emily (Lily Collins) cuts her own bangs in the Season Three premiere, and it was a case of life imitating art since Collins had recently made her own strand switch-up in real life. From there on out, Fitoussi was feeling particularly inspired about the show's creative direction.
"When Lily cut her bangs, for me that was a trigger because it reminded me immediately of the pop icons of the '60s in France," she explains. "So I said, 'OK, the '60s can be a very nice starting point on Lily."
This beautiful coat was one of Fitoussi's favorite style moments of Season 3.Stéphanie Branchu / Netflix
On her favorite outfit from Season 3
During the Season Three premiere, Emily has a nightmare that she falls off the Eiffel Tower while wearing a stunning pink coat. After reading the script, Fitoussi knew she wanted something flowy that would look beautiful while the character made her descent from the top of the tower.
"I wanted her to fall like a bird of paradise," the costume designer explains.
Alas, Fitoussi couldn't find the perfect piece of outerwear so she decided to design it herself. She had the feathered frock dyed the same colors as the dress Emily was wearing underneath — pink and red.
"For me, it was a way to say goodbye to the color pink on Emily with the fall. It was like, 'Ok, bye bye birdie' and 'Bye bye pink color,'" she said.
Lily Collins as Emily and Ashley Park as Mindy.Stéphanie Branchu / Netflix
On Emily's style evolution
Ever since moving to Paris for work, Emily has turned the city's streets into her very own runway. We've seen subtle changes in her style since Season One and now, the marketing professional has really come into her own in Season Three.
"She's evolving because she's also growing up in her job and in her personal life," she said.
Looking ahead, we can expect Emily's sense of style to continue to evolve, but one word Fitoussi will never use to describe her is "basic."
"We're not watching this show to see a hoodie, a pair of jeans and sneakers," the costume designer explained.
Emily's closet is full of some incredible pieces. / Netflix
On the million dollar question: How does Emily afford her killer wardrobe?
While watching the show, it's hard not to wonder how Emily can pay for her fabulous wardrobe. And while Fitoussi certainly understands people's curiosity, she would prefer it if everyone would just sit back and enjoy the fashion show.
Luckily, most fans do enjoy the escapism. One even messaged Fitoussi and compared "Emily in Paris" to "Sex and the City," saying that she prefers the approach that "Emily in Paris" takes with fashion.
"In 'Sex and the City,' many times you heard Sarah Jessica Parker say, 'Oh, I'm broke, I don't have money.' And she said at the end it was a little bit annoying because we couldn't enjoy the fashion as it is," the costume designer explained.
"She said, 'What I really liked with ('Emily in Paris') was that money doesn't exist in a way and we assume that completely, even if we know that we are living in very complicated times," she continued.
Emily's French coworkers, Luc (Bruno Gouery) and Julien (Samuel Arnold) are also quite stylish. Netflix
On the impact that 'Emily in Paris' has had on French fashion
French women are often known for dressing in neutral colors, but Fitoussi is particularly thrilled to see how receptive many of them have been to Emily's bold color palette.
"Many people send me messages telling me how good it is to be dressed in color because we were our own jailhouse. We prohibited ourselves to be dressed in color because we are Parisian. But who decided for us that because we are Parisian we couldn't be dressed in color?" the sartorial savant said.
Emily occasionally takes style cues from her French boss, Sylvie. / Netflix
On whether or not Emily's style has gotten more French over the seasons
Emily's wardrobe is full of bold colors, fun prints and tons of texture. We don't often see her in neutrals (a Parisian staple), but everyone's favorite marketing professional has adopted her own unique take on the French aesthetic, according to Fitoussi.
"She makes Parisian style her own and I think (her approach is) much more interesting," she said.
Bold colors are fun prints are staples in Emily's wardrobe.Netflix
The style savant loves that Emily's bold, eclectic outfits help her stand out from her French colleagues, but she has also learned a thing or two from her Parisian boss Sylvie (Philippine Leroy-Beaulieu), and vice versa. At times, we see Sylvie wearing a bit more color and Emily toning down her outfits a touch depending on the situation.
"It was very fun for me to design the two of them because one becomes, at least a little bit, what the other one is," she said.
Mixing and matching prints with solids can make a powerful impact. / Netflix
Of course, Emily's bold color palette isn't going anywhere and Fitoussi is glad about that.
"She shouldn't give it up. If she gives it up, the show doesn't exist anymore. We're not here to see Emily with a white T-shirt and a pair of jeans, even if they're the perfect jeans and even if she looks like a French icon," she said.
Pretty plaids are a mainstay in Emily's closet. / Netflix
On Emily's ringard style
Early on in the series, Emily meets a French designer named Pierre Cadault (Jean-Christophe Bouvet) and he's horrified by her ringard (corny) bag charm. Later on, Emily confronts him and owns her sense of style. After that moment, Fitoussi leaned in to the concept of being ringard and she said it's something we can all learn from.
"I think we are all the little ringard of someone. I'm probably a ringard for many people who hate what I'm doing. It doesn't matter," she explained.
Emily and Mindy are two stylish pals. / Netflix
The costume designer loved that Emily was bold enough to stay true to her unique sense of style. "It allows people to not follow the rules and to find pieces that are flattering on their body, even if it's not something trendy," she said.
To Fitoussi, confidence is beauty and having the courage to be true to yourself is admirable. "If you like glitter at 10 a.m., do it!" she said.
Black and white and oh so chic.Marie Etchegoyen / Netflix
On this stunning style moment from Season 3
During a work event in Season 3, Emily rocks an eye-catching black and white jacket with bold, structural details on the sleeves. When Fitoussi first saw the fancy frock, she originally thought it would be perfect for the character Mindy (Ashley Park).
"When I read the script and saw there was this Pierre Cadault tribute and I knew the location, I (knew) I wanted to have something with proportion because I knew that Sylvie and Madeline (Kate Walsh) would be wearing this tremendous dress with very huge proportions," she explained. "I have always loved oversized proportion. I think it's beautiful, it's bold."
Lily Collins is equally stylish in real life as she is on TV. / Netflix
On Lily Collins' style
In real life, Collins' sense of style is pretty different from Emily's, but she's always game to try whatever Fitoussi has in mind and the costume designer described the actor as "fearless."
"When you see her in real life, she likes comfy clothes like hoodies and Birkenstocks. And 10 minutes after, she can walk with confidence on chunky shoes (on set). I admire her for that because she never says never," she said.
Emily in Paris. (L to R) Ashley Park as Mindy, Lily Collins as Emily, Camille Razat as Camille in episode 301 of Emily in Paris. Cr. Stéphanie Branchu/Netflix © 2022 / Netflix
The costume designer enjoys doing fittings with Collins and said the actor is very quick to understand the concept of whatever outfit Fitoussi is envisioning for any given scene.
"She has a very clever mind," she said.
Chrissy Callahan
Chrissy Callahan covers a range of topics for TODAY.com, including fashion, beauty, pop culture and food. In her free time, she enjoys traveling, watching bad reality TV and consuming copious amounts of cookie dough.

redpajama_set_name: RedPajamaCommonCrawl


The TAKE ACTION! Network Blog: Praise Report - THROUGH THE FIRE!
Praise Report - THROUGH THE FIRE!
HALLELUJAH! GLORY BE TO GOD! THANK YOU JESUS!
The house next door was destroyed, smoke filled my apartment building and my apartment, our smoke detectors did not go off, but NO ONE was hurt or injured.
PRAISE GOD! THANK YOU JESUS!
No one was hurt! Possessions were lost. Clothes and belongings in my apartment are smoke stained, but everyone walked away with their lives - no one even needed emergency medic services!
Of course we were all shaken. I had to sleep at a neighbor's house two doors down until the smoke smell decreased enough for me to come back into my apartment and start the cleanup and airing out of the apartment. I think for the first day or so on Saturday and Sunday I was in somewhat of shock because I realized that I could have just died in my sleep from smoke inhalation, but GOD!
I am so thankful for my life? There are so many other reasons why I should be and am thankful, but I this weekend I experienced thankfulness on a whole new level.
"He who dwells in the shelter of the Most High will rest in the shadow of the Almighty. I will say of the Lord, 'He is my refuge and my fortress, my God, in whom I trust.' Surely He will save you from the fowler's snare and from the deadly pestilence. He will cover you with his feathers, and under His wings you will find refuge; his faithfulness will be your shield and rampart, You will not fear the terror of night, nor the arrow that flies by day, nor the pestilence that stalks in the darkness, nor the plague that destroys at midday. A thousand may fall at your side, ten thousand at your right hand, but it will not come near you. You will only observe with your eyes and see the punishment of the wicked.
If you make the Most High your dwelling - even the Lord, who is my refuge, then no harm will befall you, no disaster will come near your tent for he will command his angels concerning you to guard you in all your ways; they will lift you up in their hands so that you will not strike your foot against a stone. You will tread upon the lion and the cobra; you will trample the great lion and the serpent.
I believe it and receive it! HALLELUJAH!


text:
Home ▸ GUMC Stories ▸ Harold P. Freeman, Champion of the Underserved, Receives Top Honors at 10th Annual Convocation
Harold P. Freeman, Champion of the Underserved, Receives Top Honors at 10th Annual Convocation
Posted in GUMC Stories | Tagged university awards
November 21, 2017 — With a grand processional entrance by Georgetown University Medical Center faculty in full, colorful regalia, the GUMC Fall Convocation began its 10th annual awards ceremony in front of a packed house in the Research Building auditorium on November 16.
As in years past, the highlight of Convocation was the bestowing of the Cura Personalis Award that recognizes a health professional who has made outstanding contributions to human health guided by compassion and service, and the recipient's keynote address. This year's recipient was Harold P. Freeman, MD, a champion of the poor and underserved.
As reflected in the award recipient, the theme for both the awards ceremony and the colloquium (new window) held earlier in the day was health disparities, an affirmed priority area for GUMC.
Groundbreaking efforts
"Dr. Freeman has devoted his life to advancing health and eliminating health disparities in vulnerable and underserved communities across our country," said Edward B. Healton, MD, MPH, executive vice president for health sciences at GUMC and executive dean of School of Medicine.
"I have admired his work over the entire span of his long career, and so it is a very special pleasure for me to stand with Dr. Freeman on this podium today," he added.
Freeman's most important work has included groundbreaking efforts in reducing health inequities, including the development and implementation of the concept known as patient navigation.
Inspiring ancestors
During his keynote address, Freeman, a surgical oncologist who spent his professional career focused on the underserved in Harlem, reflected on his ancestors, enslaved since before the signing of the Declaration of Independence, he said.
His ancestors eventually "bought" their freedom.
"If my ancestors could do that, I was challenged to do as much as I could and more if I could."
Freeman stayed in Washington to attend medical school at Howard University. He began his career at Harlem Hospital where he witnessed health disparities and injustice towards his poor and black cancer patients in the breast clinic that he ran.
"This changed my life …"
"Many of those women were coming in too late," Freeman explained. "They were all poor. They were all black. Some of the women had visible masses. You don't have to touch it – it's there. Some of the masses were bleeding when I first saw them. This changed my life. … I had to do something more. I could not accept that people were coming in this late and dying."
After talking to the women, he learned that for many, this had not been their first visit to a doctor. Many had been to the emergency room and turned away because a breast mass didn't represent an emergency and they didn't have insurance. But Freeman recalled the women explaining they had other concerns, like food, housing and crime in the streets.
Freeman said he knew he had to do more. "It didn't make moral sense to me," he said.
Opened Free Clinics
As director of surgery at Harlem Hospital, he opened a free Saturday screening clinic at Harlem Hospital in 1979, and he founded the Breast Examination Center of Harlem, a program that continues today at Memorial Sloan Kettering to provide free breast cancer screenings.
"I solved the challenge of screening people but that wasn't enough, because some of the women are lost after a finding [on their mammogram]," he said. "And so I knew I had not really solved the problem."
Freeman became president of the American Cancer Society in 1988. Believing that poverty more than race played a role in the outcomes of his cancer patients, Freeman launched a listening tour across the country to learn about the challenges faced by poor people with cancer, whether in California, New Jersey, or Texas.
He heard the same message repeatedly from poor people of all races. "They said 'We meet barriers when we attempt to get into, and through, this very, very complex health care system,' " Freeman recalled. "I thought, if people meet barriers, maybe we can navigate them."
And with that, Freeman's most innovative idea had taken root.
The Birth of Patient Navigation
Freeman created and implemented the concept of "patient navigation" in 1999. It is intended to match a trained individual with a cancer patient to guide the individual through the complexities of the health care system – from screening to diagnosis and through the completion of treatment.
Freeman first added navigation to the free screening clinic in Harlem, and it was a stunning success. He and his colleagues published data in 1995 describing its impact on a patient's outcome.
"We had changed the five-year survival from 39 percent, before screening and intervention, to 70 percent," Freeman told the audience, which erupted in applause.
The data were presented to Congress, and in 2005, George W. Bush signed the National Patient Navigation Act.
Navigation Institute
Freeman went on to found the Harold P. Freeman Navigation Institute, a program that provides navigator training. His approached has been embraced by organizations around the country including the Capital Breast Cancer Program, a Georgetown Lombardi Comprehensive Cancer Center program in southeast Washington.
Freeman also was founding director of the National Cancer Institute's Center to Reduce Health Disparities. He is a member of the National Academy of Medicine and has received a Lasker Award for public service — one of the numerous accolades Freeman has garnered in his lifetime.
Unwavering Dedication
At the conclusion of his address, Georgetown University President John J. DeGioia read the citation for the Cura Personalis Award.
"Georgetown University Medical Center celebrates the accomplishments of an individual who has worked tirelessly to reduce health disparities by developing programs to eliminate barriers that often prevent patients from accessing critical screening and treatment services.
"For his unwavering dedication to the underserved, for his achievements as an innovator, and for his service to national organizations through which he worked to address health disparities, Georgetown University Medical Center presents Harold P. Freeman MD, with its highest honor, the Cura Personalis Award."


text:
Accenture Interactive buys design shop Insitum to firm up Latin American presence
Accenture Interactive, the digital marketing services division of Accenture, acquired the Mexico-based design and strategic research firm Insitum for an undisclosed sum, according to a press release.
Insitum, founded in 2003, will support Accenture Interactive's existing design unit, Fjord. Insitum has more than 200 employees and is known for work with both local and global clients, including Bancolombia, BBVA, CEMEX, Grupo Bimbo, Roche and Uber. The deal comes as Accenture looks to build out a stronger presence in Latin American countries like Argentina, Colombia, Peru, Brazil and especially Mexico.
Accenture Interactive recently acquired two other Latin American agencies: Sao Paulo-based content marketing specialist New Content and the Brazilian digital marketing and commerce shop AD.Dialeto. Shackleton, which is based in Spain, also gave the consultancy a larger presence in Chile when it was bought by Accenture Interactive in April.
Accenture Interactive continues an acquisition streak with the purchase of Insitum, which joins more than two dozen marketing firms that have folded into the consultancy's global network over the past decade. The deal indicates where Accenture Interactive is eyeing new areas of growth, extending a series of Latin America-focused acquisitions that span digital marketing, commerce and design.
Insitum's support of Fjord links back to the roots of Accenture Interactive's M&A strategy. The division made one of its first big waves in the advertising world when it bought Fjord in 2013.
Fjord has grown in the time since, expanding a presence both stateside — the company formed in London in 2001, but has a significant U.S. presence — and abroad to become one of Accenture Interactive's centerpiece brands. With Insitum, the unit and Accenture Interactive's other Latin American operations could glean more local knowledge and talent, as well as connections to deep-pocketed clients like Uber.
"Today's problems can no longer be solved with yesterday's solutions," Baiju Shah, Accenture Interactive's chief strategy officer and and a global co-lead at Fjord, said in a statement.
"To tackle these challenges, clients must be agile and focus on blending a human-centered design approach with an analytics-led strategy to keep ahead of the unprecedented pace of change and ambiguity we live in," Shah said.
Localization, or the ability to tailor work based on local cultures and interests, remains a pain point for many marketers. Just 10% of global marketing leaders recently surveyed by the CMO Council reported feeling "exceedingly confident" in reaching customer engagement and revenue goals on this front. Operational structures, in particular, were found to hamper customer relationships.
Accenture Interactive's regional specialization and data-driven marketing services have appeared to be a boon while the group's largest competitors, namely major ad holding groups like WPP and Omnicom, have seen growth slow. Discussing earnings results earlier this summer, executives forecast that Accenture Interactive could increase revenue 20% this year to surpass $10 billion.
Some of Accenture Interactive's acquisitions of late have raised eyebrows. In April, it purchased Droga5, a decorated creative firm, marking the biggest deal in its 10-year history. Some industry analysts have wondered how the agency, which was proudly independent for years, will preserve its culture under the new ownership.
Accenture Interactive Accenture Interactive to Acquire Leading Design and Innovation Firm, INSITUM
Digiday WPP and Omnicom take a stand against Accenture-led audits
Filed Under: Analytics Marketing Tech
Kendall Davis/Industry Dive/Marketing Dive, data from Marc Nozell
How a Joe Biden presidency could impact brand marketing
Historically risk-averse marketers embraced causes from immigration to environmentalism under President Donald Trump. Lingering division means the industry shift will be less radical this go-around.
By Peter Adams • Nov. 09, 2020
Latest in Analytics
ISpot buys Ace Metrix to expand analytics for TV and CTV ads
What to know about second-party data as marketers tilt toward collaboration
Garnier tweaks social strategy to guide at-home hair colorists during pandemic
Ads on Facebook may not be that important for brands after all, Forrester data shows
By Ben Unglesbee • Jan. 13, 2021


Hardcell is a modified version of Freecell Solitaire. Freecell is a pretty easy game to win, and Hardcell is much more difficult for those looking for a challenge. In Hardcell, you are only allowed to pick up the top card in the column. Also there are 20 cards turned face down that you can not look at until you have removed all of the cards overlapping them in the column.