@SuppressWarnings("deprecation")
  public Vertex createVertex(JobConf conf, BaseWork workUnit, Path scratchDir,
      TezWork tezWork, Map<String, LocalResource> localResources) throws Exception {

    Vertex vertex;
    // simply dispatch the call to the right method for the actual (sub-) type of
    // BaseWork.
    VertexType vertexType = tezWork.getVertexType(workUnit);
    if (workUnit instanceof MapWork) {
      vertex = createVertexFromMapWork(
          conf, (MapWork) workUnit, scratchDir, vertexType);
    } else if (workUnit instanceof ReduceWork) {
      vertex = createVertexFromReduceWork(conf, (ReduceWork) workUnit, scratchDir);
    } else if (workUnit instanceof MergeJoinWork) {
      vertex = createVertexFromMergeWork(
          conf, (MergeJoinWork) workUnit, scratchDir, vertexType);
      // set VertexManagerPlugin if whether it's a cross product destination vertex
      List<String> crossProductSources = new ArrayList<>();
      for (BaseWork parentWork : tezWork.getParents(workUnit)) {
        if (tezWork.getEdgeType(parentWork, workUnit) == EdgeType.XPROD_EDGE) {
          crossProductSources.add(parentWork.getName());
        }
      }

      if (!crossProductSources.isEmpty()) {
        CartesianProductConfig cpConfig = new CartesianProductConfig(crossProductSources);
        vertex.setVertexManagerPlugin(
          VertexManagerPluginDescriptor.create(CartesianProductVertexManager.class.getName())
            .setUserPayload(cpConfig.toUserPayload(new TezConfiguration(conf))));
        // parallelism shouldn't be set for cartesian product vertex
      }
    } else {
      // something is seriously wrong if this is happening
      throw new HiveException(ErrorMsg.GENERIC_ERROR.getErrorCodedMsg());
    }
    VertexExecutionContext vertexExecutionContext = createVertexExecutionContext(workUnit);
    vertex.addTaskLocalFiles(localResources);
    vertex.setTaskLaunchCmdOpts(getContainerJavaOpts(conf));
    vertex.setExecutionContext(vertexExecutionContext);

    // initialize stats publisher if necessary
    if (workUnit.isGatheringStats()) {
      StatsPublisher statsPublisher;
      StatsFactory factory = StatsFactory.newFactory(conf);
      if (factory != null) {
        StatsCollectionContext sCntxt = new StatsCollectionContext(conf);
        sCntxt.setStatsTmpDirs(Utilities.getStatsTmpDirs(workUnit, conf));
        statsPublisher = factory.getStatsPublisher();
        if (!statsPublisher.init(sCntxt)) { // creating stats table if not exists
          if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_STATS_RELIABLE)) {
            throw
              new HiveException(ErrorMsg.STATSPUBLISHER_INITIALIZATION_ERROR.getErrorCodedMsg());
          }
        }
      }
    }

    final Class outputKlass;
    if (HiveOutputFormatImpl.class.getName().equals(conf.get("mapred.output.format.class"))) {
      // Hive uses this output format, when it is going to write all its data through FS operator
      outputKlass = NullMROutput.class;
    } else {
      outputKlass = MROutput.class;
    }

    // If there is a fileSink add a DataSink to the vertex
    boolean hasFileSink = workUnit.getAllOperators().stream().anyMatch(o -> o instanceof FileSinkOperator);
    // final vertices need to have at least one output
    boolean endVertex = tezWork.getLeaves().contains(workUnit);
    if (endVertex || hasFileSink) {
      OutputCommitterDescriptor ocd = null;
      String committer = HiveConf.getVar(conf, ConfVars.TEZ_MAPREDUCE_OUTPUT_COMMITTER);
      if (committer != null && !committer.isEmpty()) {
        ocd = OutputCommitterDescriptor.create(committer);
      }
      vertex.addDataSink("out_"+workUnit.getName(), new DataSinkDescriptor(
          OutputDescriptor.create(outputKlass.getName())
          .setUserPayload(vertex.getProcessorDescriptor().getUserPayload()), ocd, null));
    }

    return vertex;
  }
Create a vertex from a given work object.

@param conf JobConf to be used to this execution unit
@param workUnit The instance of BaseWork representing the actual work to be performed
by this vertex.
@param scratchDir HDFS scratch dir for this execution unit.
@return Vertex
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java