DAG build(JobConf conf, TezWork tezWork, Path scratchDir, Context ctx,
      Map<String, LocalResource> vertexResources) throws Exception {

    perfLogger.perfLogBegin(CLASS_NAME, PerfLogger.TEZ_BUILD_DAG);

    // getAllWork returns a topologically sorted list, which we use to make
    // sure that vertices are created before they are used in edges.
    List<BaseWork> topologicalWorkList = tezWork.getAllWork();
    Collections.reverse(topologicalWorkList);

    // the name of the dag is what is displayed in the AM/Job UI
    String dagName = utils.createDagName(conf, queryPlan);

    LOG.info("Dag name: {}", dagName);
    DAG dag = DAG.create(dagName);

    // set some info for the query
    JSONObject json = new JSONObject(new LinkedHashMap<>()).put("context", "Hive")
        .put("description", ctx.getCmd());
    String dagInfo = json.toString();

    String queryId = HiveConf.getVar(conf, HiveConf.ConfVars.HIVE_QUERY_ID);
    dag.setConf(HiveConf.ConfVars.HIVE_QUERY_ID.varname, queryId);

    LOG.debug("DagInfo: {}", dagInfo);

    TezConfigurationFactory.addProgrammaticallyAddedTezOptsToDagConf(dag.getDagConf(), conf);
    dag.setDAGInfo(dagInfo);

    dag.setCredentials(conf.getCredentials());
    setAccessControlsForCurrentUser(dag, queryPlan.getQueryId(), conf);

    for (BaseWork workUnit: topologicalWorkList) {

      // translate work to vertex
      perfLogger.perfLogBegin(CLASS_NAME, PerfLogger.TEZ_CREATE_VERTEX + workUnit.getName());

      if (workUnit instanceof UnionWork) {
        // Special case for unions. These items translate to VertexGroups

        List<BaseWork> unionWorkItems = new LinkedList<BaseWork>();
        List<BaseWork> children = new LinkedList<BaseWork>();

        // split the children into vertices that make up the union and vertices that are
        // proper children of the union
        for (BaseWork v: tezWork.getChildren(workUnit)) {
          EdgeType type = tezWork.getEdgeProperty(workUnit, v).getEdgeType();
          if (type == EdgeType.CONTAINS) {
            unionWorkItems.add(v);
          } else {
            children.add(v);
          }
        }
        JobConf parentConf = workToConf.get(unionWorkItems.get(0));
        checkOutputSpec(workUnit, parentConf);

        // create VertexGroup
        Vertex[] vertexArray = new Vertex[unionWorkItems.size()];

        int i = 0;
        for (BaseWork v: unionWorkItems) {
          vertexArray[i++] = workToVertex.get(v);
        }
        VertexGroup group = dag.createVertexGroup(workUnit.getName(), vertexArray);

        // For a vertex group, all Outputs use the same Key-class, Val-class and partitioner.
        // Pick any one source vertex to figure out the Edge configuration.

        // now hook up the children
        for (BaseWork v: children) {
          // finally we can create the grouped edge
          GroupInputEdge e = utils.createEdge(group, parentConf,
               workToVertex.get(v), tezWork.getEdgeProperty(workUnit, v), v, tezWork);

          dag.addEdge(e);
        }
      } else {
        // Regular vertices
        JobConf wxConf = utils.initializeVertexConf(conf, ctx, workUnit);
        checkOutputSpec(workUnit, wxConf);
        Vertex wx = utils.createVertex(wxConf, workUnit, scratchDir,
            tezWork, vertexResources);
        if (tezWork.getChildren(workUnit).size() > 1) {
          String tezRuntimeSortMb = wxConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB);
          int originalValue = 0;
          if(tezRuntimeSortMb == null) {
            originalValue = TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB_DEFAULT;
          } else {
            originalValue = Integer.valueOf(tezRuntimeSortMb);
          }
          int newValue = originalValue / tezWork.getChildren(workUnit).size();
          wxConf.set(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, Integer.toString(newValue));
          LOG.info("Modified " + TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB + " to " + newValue);
        }
        if (workUnit.getReservedMemoryMB() > 0) {
          // If reversedMemoryMB is set, make memory allocation fraction adjustment as needed
          double frac = DagUtils.adjustMemoryReserveFraction(workUnit.getReservedMemoryMB(), super.conf);
          LOG.info("Setting " + TEZ_MEMORY_RESERVE_FRACTION + " to " + frac);
          wx.setConf(TEZ_MEMORY_RESERVE_FRACTION, Double.toString(frac));
        } // Otherwise just leave it up to Tez to decide how much memory to allocate
        dag.addVertex(wx);
        utils.addCredentials(workUnit, dag, conf);
        perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.TEZ_CREATE_VERTEX + workUnit.getName());
        workToVertex.put(workUnit, wx);
        workToConf.put(workUnit, wxConf);

        // add all dependencies (i.e.: edges) to the graph
        for (BaseWork v: tezWork.getChildren(workUnit)) {
          assert workToVertex.containsKey(v);
          Edge e = null;

          TezEdgeProperty edgeProp = tezWork.getEdgeProperty(workUnit, v);
          e = utils.createEdge(wxConf, wx, workToVertex.get(v), edgeProp, v, tezWork);
          dag.addEdge(e);
        }
      }
    }
    // Clear the work map after build. TODO: remove caching instead?
    Utilities.clearWorkMap(conf);
    perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.TEZ_BUILD_DAG);
    return dag;
  }
Ensures that the Tez Session is open and the AM has all necessary jars configured.
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java