JobGraph createJobGraph(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc) {
    JobGraph jobGraph = new JobGraph(config, appDesc);
    // Source streams contain both input and intermediate streams.
    Set<StreamSpec> sourceStreams = getStreamSpecs(appDesc.getInputStreamIds(), streamConfig);
    // Sink streams contain both output and intermediate streams.
    Set<StreamSpec> sinkStreams = getStreamSpecs(appDesc.getOutputStreamIds(), streamConfig);

    Set<StreamSpec> intermediateStreams = Sets.intersection(sourceStreams, sinkStreams);
    Set<StreamSpec> inputStreams = Sets.difference(sourceStreams, intermediateStreams);
    Set<StreamSpec> outputStreams = Sets.difference(sinkStreams, intermediateStreams);

    Set<TableDescriptor> tables = appDesc.getTableDescriptors();

    // Generate job.id and job.name configs from app.id and app.name if defined
    MapConfig generatedJobConfigs = JobPlanner.generateSingleJobConfig(config);
    String jobName = generatedJobConfigs.get(JobConfig.JOB_NAME);
    String jobId = generatedJobConfigs.get(JobConfig.JOB_ID, "1");

    // For this phase, we have a single job node for the whole DAG
    JobNode node = jobGraph.getOrCreateJobNode(jobName, jobId);

    // Add input streams
    inputStreams.forEach(spec -> jobGraph.addInputStream(spec, node));

    // Add output streams
    outputStreams.forEach(spec -> jobGraph.addOutputStream(spec, node));

    // Add intermediate streams
    intermediateStreams.forEach(spec -> jobGraph.addIntermediateStream(spec, node, node));

    // Add tables
    for (TableDescriptor table : tables) {
      jobGraph.addTable(table, node);
      // Add side-input streams (if any)
      if (table instanceof LocalTableDescriptor) {
        LocalTableDescriptor localTable = (LocalTableDescriptor) table;
        Iterable<String> sideInputs = ListUtils.emptyIfNull(localTable.getSideInputs());
        for (String sideInput : sideInputs) {
          jobGraph.addSideInputStream(getStreamSpec(sideInput, streamConfig));
        }
      }
    }

    if (!LegacyTaskApplication.class.isAssignableFrom(appDesc.getAppClass())) {
      // skip the validation when input streamIds are empty. This is only possible for LegacyTaskApplication
      jobGraph.validate();
    }

    return jobGraph;
  }
Creates the physical graph from {@link ApplicationDescriptorImpl}
samza-core/src/main/java/org/apache/samza/execution/ExecutionPlanner.java