5-10transformation中部分函数使用

目录

1. transformation函数之map(),filter()函数的简单应用：

2. union

3.Split与Select

1. transformation函数之map(),filter()函数的简单应用：

object DataStreamTransformationAPP {def main(args: Array[String]): Unit = {val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironmentfilterFunction(env)env.execute("DataStreamTransformationAPP")}def filterFunction(env: StreamExecutionEnvironment): Unit ={val data: DataStream[Long] = env.addSource(new CustomerNonParallesourceFunction)data.map(x=>{println("received:"+x)  //此处为接收到的数据，即未处理的数据x}).filter(_%2==0)//此处过滤.print().setParallelism(1)}
}

2. union

dataStream.union(otherStream1, otherStream2, ...);

Union of two or more data streams creating a new stream containing all the elements from all the streams.

Note: If you union a data stream with itself you will get each element twice in the resulting stream.：如果你union数据流和它本身，你会的到这个数据流两次

def unionFunction(env: StreamExecutionEnvironment): Unit ={val data1: DataStream[Long] = env.addSource(new CustomerNonParallesourceFunction)val data2: DataStream[Long] = env.addSource(new CustomerNonParallesourceFunction)data1.union(data2).print().setParallelism(1)}

3.Split与Select

1)DataStream → SplitStream:Split the stream into two or more streams according to some criterion.

val split = someDataStream.split(
  (num: Int) =>
    (num % 2) match {
      case 0 => List("even")//奇数
      case 1 => List("odd")//偶数
    }
)

2）SplitStream → DataStream：Select one or more streams from a split stream.

val even = split select "even"
val odd = split select "odd"
val all = split.select("even","odd")

  def splitSelectFunction(env: StreamExecutionEnvironment): Unit ={val data: DataStream[Long] = env.addSource(new CustomerNonParallesourceFunction)val splits: SplitStream[Long] = data.split(new OutputSelector[Long] {override def select(value: Long): lang.Iterable[String] = {val list = new util.ArrayList[String]()if (value % 2 != 0) {list.add("even")} else {list.add("odd")}list}})splits.select("odd").print().setParallelism(1)}