分布式计算 DAG1- 画猫

type StreamExecutionEnvironment struct {
  DefaultJobName     string
  TimeCharacteristic TimeCharacteristic   //这个时间标示有3个，processTime、ingestionTime、EventTime。分别表示数据处理时间、数据获取时间、数据本身的时间，也就是说在进行时间窗口计算的时候，有三种时间标示可以配置和选择，分别应对不同的处理场景。
  DefaultParallelism int
  Config             *ExecutionConfig
  Transformations   []*StreamTransformation   //记录了用户所编排的算子，使用List来维护。
  isChainingEnabled bool
  HostName           string
  Port               int
}

type StreamTransformation struct {
  Id int                   
  Name string
  OutPutType interface{}
  Parallelism int       // 这里有个并行度设置，默认是本节点cpu cores，但是最终会根据不同的算子类型设置        
  MinResources *operator.ResourceSpec   //默认配置资源，主要包括cpu cores和mem size 
  PreferredResources *operator.ResourceSpec //自定义配置
  uid string 
  slotSharingGroup string //这个主要用于表示哪些算子是属于同一个组的
  coLocationGroupKey string
  input interface{}   //该transform的上游输入算子
  operator interface{}   //该transform的具体算子
}

func main() {
  hostName,_:=os.Hostname()
  port:=12345
  env:= dataStream.NewStreamExecutionEnvironment(hostName,port) //初始化可执行环境
  env.Config.ParallelismDefault=runtime.NumCPU()   //这里配置了默认的并行读为cpu cores 
  data:="dsdsadadasdsdsda"                                     
  dataSource:=env.FromData(data,newReadData())   //算子编排从这里开始，传入了一个测试数据data和自定义的处理函数
  dataSource.FlatMap(newMapData()).KeyBy(newKeySelector()).FlatMap(newMapData()).PrintData(newWriteData())
  env.Execute("first job !")
}

const (
  ALWAYS = iota   上下游算子都可以链接该算子
  NEVER           上下游算子不可以链接该算子
  HEAD             标示该算子是DAG的头节点，也就是说只能允许其下游节点进行链接
)

type DataStream struct {
  Environment *StreamExecutionEnvironment
  Transformation *StreamTransformation
}

streamGraph:=env.getStreamGraph(jobName)   //这里首先会生成原始图，也就是streamGraph
jobGraphGenerator:=NewJobGraphGenerator(streamGraph,"") //然后在原始图的基础上，会进行图优化、配置算子中间结果数据集和算子通信的通道。
jobGraph:=jobGraphGenerator.CreateJobGraph() //生成jobGraph
jobGraph.JobName=jobName

func (env *StreamExecutionEnvironment) getStreamGraph(jobName string) *StreamGraph {
  return env.getStreamGraphGenerator().Generate()
}

type StreamGraphGenerator struct {
  JobName             string
  Transformations     []*StreamTransformation   //其实这里的核心关键就是，生成图所需的transformation
  ExecutionConfig     *ExecutionConfig
  Chaining             bool
  IsSlotSharingEnabled bool
  AlreadyTransformed   map[*StreamTransformation][]int   //另外一个，维护在遍历过程中，维护已经访问过的transformation
  StreamGraph         *StreamGraph   //还有就是图
  TimeCharacteristic   TimeCharacteristic
}



type StreamGraph struct {
  JobName               string
  ExecutionConfig       *ExecutionConfig
  chaining             bool
  timeCharacteristic   TimeCharacteristic
  StreamNodes           map[int]*StreamNode   对于图这里会维护图的节点信息，而每个节点中会维护该节点的出边和入边
  Sources               map[int]bool         当然，还有根据算子类型判断出来的，source和sink的transformation id 
  Sinks                 map[int]bool
  virtualPartitionNodes map[int]*StreamPartitioner
}

transformIds,ok:=generator.AlreadyTransformed[trans]   //这里首先就会判断，这个transformation是不是遍历过了，如果没有就返回了。
if ok {
  return transformIds
}
if trans.Parallelism <= 0 {   //又再次判断是不是配置了并行度
  trans.Parallelism=generator.ExecutionConfig.ParallelismDefault
}
  _,ok=reflect.TypeOf(trans.operator).MethodByName("SourceRead")   判断这个transform的类型，这里将transform根据算子类型分了几类，这里我门看一下transformSource就可以了，因为都是大同小异的。
  if ok {
      transformIds=generator.transformSource(trans)
}
  _,ok=reflect.TypeOf(trans.operator).MethodByName("SinkWrite")
  if ok {
      transformIds=generator.transformSink(trans)
}
  _,ok=reflect.TypeOf(trans.operator).MethodByName("Reduce")
  if ok {
  transformIds=generator.transformReduce(trans)
}
  _,ok=reflect.TypeOf(trans.operator).MethodByName("Map")
  if ok {
  transformIds=generator.transformMap(trans)
}
  _,ok=reflect.TypeOf(trans.operator).MethodByName("GetKey")
  if ok {
  transformIds=generator.transformPartition(trans)
}
  _,ok=generator.AlreadyTransformed[trans]
  if !ok {
      generator.AlreadyTransformed[trans]=make([]int,0)   //将遍历过的transform添加到该map list中。
      generator.AlreadyTransformed[trans]=append(generator.AlreadyTransformed[trans],transformIds...)
}
  if trans.uid !="" {
      generator.StreamGraph.SetTransformationUID(trans.Id,trans.uid)
}
  if trans.MinResources !=nil && trans.PreferredResources !=nil {
      generator.StreamGraph.SetResources(trans.Id,trans.MinResources,trans.PreferredResources)
}
return transformIds

//根据节点id,获取上下游节点
upStreamNode:=sg.GetStreamNode(upStreamVertexId)
downStreamNode:=sg.GetStreamNode(downStreamVertexId)
//创建上下游节点的边
edge:=NewStreamEdge(upStreamNode,downStreamNode,outputNames,partitioner)
//为下游节点添加入边
sg.GetStreamNode(edge.TargetId).InEdges=append(sg.GetStreamNode(edge.TargetId).InEdges,edge)
//为上游节点添加出边
sg.GetStreamNode(edge.SourceId).OutEdges=append(sg.GetStreamNode(edge.SourceId).OutEdges,edge)