public <R> SingleOutputStreamOperator<R> map
 
(MapFunction<T, R> mapper) {
  // 通過java reflection抽出mapper的返回值型別
  TypeInformation<R> outType = TypeExtractor.getMapReturnTypes(clean(mapper), getType(),
      Utils.getCallLocationName(), true);

  // 返回一個新的DataStream，SteramMap 為 StreamOperator 的實現類
  return transform("Map", outType, new StreamMap<>(clean(mapper)));
}

public <R> SingleOutputStreamOperator<R> transform(String operatorName, TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {
  // read the output type of the input Transform to coax out errors about MissingTypeInfo
  transformation.getOutputType();

  // 新的transformation會連線上當前DataStream中的transformation，從而構建成一棵樹
  OneInputTransformation<T, R> resultTransform = new OneInputTransformation<>(
      this.transformation,
      operatorName,
      operator,
      outTypeInfo,
      environment.getParallelism());

  @SuppressWarnings({ "unchecked", "rawtypes" })
  SingleOutputStreamOperator<R> returnStream = new SingleOutputStreamOperator(environment, resultTransform);

  // 所有的transformation都會存到 env 中，呼叫execute時遍歷該list生成StreamGraph
  getExecutionEnvironment().addOperator(resultTransform);

  return returnStream;
}

// 構造 StreamGraph 入口函式
public static StreamGraph generate(StreamExecutionEnvironment env, List<StreamTransformation<?>> transformations) {
    return new StreamGraphGenerator(env).generateInternal(transformations);
}

// 自底向上（sink->source）對轉換樹的每個transformation進行轉換。
private StreamGraph generateInternal(List<StreamTransformation<?>> transformations) {
  for (StreamTransformation<?> transformation: transformations) {
    transform(transformation);
  }
  return streamGraph;
}

// 對具體的一個transformation進行轉換，轉換成 StreamGraph 中的 StreamNode 和 StreamEdge
// 返回值為該transform的id集合，通常大小為1個（除FeedbackTransformation）
private Collection<Integer> transform(StreamTransformation<?> transform) {  
  // 跳過已經轉換過的transformation
  if (alreadyTransformed.containsKey(transform)) {
    return alreadyTransformed.get(transform);
  }

  LOG.debug("Transforming " + transform);

  // 為了觸發 MissingTypeInfo 的異常
  transform.getOutputType();

  Collection<Integer> transformedIds;
  if (transform instanceof OneInputTransformation<?, ?>) {
    transformedIds = transformOnInputTransform((OneInputTransformation<?, ?>) transform);
  } else if (transform instanceof TwoInputTransformation<?, ?, ?>) {
    transformedIds = transformTwoInputTransform((TwoInputTransformation<?, ?, ?>) transform);
  } else if (transform instanceof SourceTransformation<?>) {
    transformedIds = transformSource((SourceTransformation<?>) transform);
  } else if (transform instanceof SinkTransformation<?>) {
    transformedIds = transformSink((SinkTransformation<?>) transform);
  } else if (transform instanceof UnionTransformation<?>) {
    transformedIds = transformUnion((UnionTransformation<?>) transform);
  } else if (transform instanceof SplitTransformation<?>) {
    transformedIds = transformSplit((SplitTransformation<?>) transform);
  } else if (transform instanceof SelectTransformation<?>) {
    transformedIds = transformSelect((SelectTransformation<?>) transform);
  } else if (transform instanceof FeedbackTransformation<?>) {
    transformedIds = transformFeedback((FeedbackTransformation<?>) transform);
  } else if (transform instanceof CoFeedbackTransformation<?>) {
    transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);
  } else if (transform instanceof PartitionTransformation<?>) {
    transformedIds = transformPartition((PartitionTransformation<?>) transform);
  } else {
    throw new IllegalStateException("Unknown transformation: " + transform);
  }

  // need this check because the iterate transformation adds itself before
  // transforming the feedback edges
  if (!alreadyTransformed.containsKey(transform)) {
    alreadyTransformed.put(transform, transformedIds);
  }

  if (transform.getBufferTimeout() > 0) {
    streamGraph.setBufferTimeout(transform.getId(), transform.getBufferTimeout());
  }
  if (transform.getUid() != null) {
    streamGraph.setTransformationId(transform.getId(), transform.getUid());
  }

  return transformedIds;
}

private <IN, OUT> Collection<Integer> transformOnInputTransform(OneInputTransformation<IN, OUT> transform) {
  // 遞迴對該transform的直接上游transform進行轉換，獲得直接上游id集合
  Collection<Integer> inputIds = transform(transform.getInput());

  // 遞迴呼叫可能已經處理過該transform了
  if (alreadyTransformed.containsKey(transform)) {
    return alreadyTransformed.get(transform);
  }

  String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), inputIds);

  // 新增 StreamNode
  streamGraph.addOperator(transform.getId(),
      slotSharingGroup,
      transform.getOperator(),
      transform.getInputType(),
      transform.getOutputType(),
      transform.getName());

  if (transform.getStateKeySelector() != null) {
    TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());
    streamGraph.setOneInputStateKey(transform.getId(), transform.getStateKeySelector(), keySerializer);
  }

  streamGraph.setParallelism(transform.getId(), transform.getParallelism());

  // 新增 StreamEdge
  for (Integer inputId: inputIds) {
    streamGraph.addEdge(inputId, transform.getId(), 0);
  }

  return Collections.singleton(transform.getId());
}

private <T> Collection<Integer> transformPartition(PartitionTransformation<T> partition) {
  StreamTransformation<T> input = partition.getInput();
  List<Integer> resultIds = new ArrayList<>();

  // 直接上游的id
  Collection<Integer> transformedIds = transform(input);
  for (Integer transformedId: transformedIds) {
    // 生成一個新的虛擬id
    int virtualId = StreamTransformation.getNewNodeId();
    // 新增一個虛擬分割槽節點，不會生成 StreamNode
    streamGraph.addVirtualPartitionNode(transformedId, virtualId, partition.getPartitioner());
    resultIds.add(virtualId);
  }

  return resultIds;
}

public void addEdge(Integer upStreamVertexID, Integer downStreamVertexID, int typeNumber) {
  addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, null, new ArrayList<String>());
}
private void addEdgeInternal(Integer upStreamVertexID,
    Integer downStreamVertexID,
    int typeNumber,
    StreamPartitioner<?> partitioner,
    List<String> outputNames) {

  // 當上遊是select時，遞迴呼叫，並傳入select資訊
  if (virtualSelectNodes.containsKey(upStreamVertexID)) {
    int virtualId = upStreamVertexID;
    // select上游的節點id
    upStreamVertexID = virtualSelectNodes.get(virtualId).f0;
    if (outputNames.isEmpty()) {
      // selections that happen downstream override earlier selections
      outputNames = virtualSelectNodes.get(virtualId).f1;
    }
    addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames);
  } 
  // 當上遊是partition時，遞迴呼叫，並傳入partitioner資訊
  else if (virtuaPartitionNodes.containsKey(upStreamVertexID)) {
    int virtualId = upStreamVertexID;
    // partition上游的節點id
    upStreamVertexID = virtuaPartitionNodes.get(virtualId).f0;
    if (partitioner == null) {
      partitioner = virtuaPartitionNodes.get(virtualId).f1;
    }
    addEdgeInternal(upStreamVertexID, downStreamVertexID, typeNumber, partitioner, outputNames);
  } else {
    // 真正構建StreamEdge
    StreamNode upstreamNode = getStreamNode(upStreamVertexID);
    StreamNode downstreamNode = getStreamNode(downStreamVertexID);

    // 未指定partitioner的話，會為其選擇 forward 或 rebalance 分割槽。
    if (partitioner == null && upstreamNode.getParallelism() == downstreamNode.getParallelism()) {
      partitioner = new ForwardPartitioner<Object>();
    } else if (partitioner == null) {
      partitioner = new RebalancePartitioner<Object>();
    }

    // 健康檢查， forward 分割槽必須要上下游的併發度一致
    if (partitioner instanceof ForwardPartitioner) {
      if (upstreamNode.getParallelism() != downstreamNode.getParallelism()) {
        throw new UnsupportedOperationException("Forward partitioning does not allow " +
            "change of parallelism. Upstream operation: " + upstreamNode + " parallelism: " + upstreamNode.getParallelism() +
            ", downstream operation: " + downstreamNode + " parallelism: " + downstreamNode.getParallelism() +
            " You must use another partitioning strategy, such as broadcast, rebalance, shuffle or global.");
      }
    }
    // 建立 StreamEdge
    StreamEdge edge = new StreamEdge(upstreamNode, downstreamNode, typeNumber, outputNames, partitioner);
    // 將該 StreamEdge 新增到上游的輸出，下游的輸入
    getStreamNode(edge.getSourceId()).addOutEdge(edge);
    getStreamNode(edge.getTargetId()).addInEdge(edge);
  }
}

DataStream<String> text = env.socketTextStream(hostName, port);
text.flatMap(new LineSplitter()).shuffle().filter(new HelloFilter()).print();