ProcessFunction

Flink DataStream API中最底层API，提供process方法，其中需要实现ProcessFunction函数类，类似的类还有KeyedProcessFunction。

ProcessFunction是一个低层次的流处理操作，允许返回所有Stream的基础构建模块:

1. 访问Event本身数据（比如：Event的时间，Event的当前Key等）
2. 管理每个key的状态State（仅在Keyed Stream中）
3. 管理定时器Timer（包括：注册定时器，删除定时器等）

总而言之，ProcessFunction是Flink最底层的API，也是功能最强大的。

案例：使用process函数，代替filter函数，实现对数据过滤操作。

xxxxxxxxxx
package com.itbaizhan.flink.scala.transformation

import org.apache.flink.api.common.functions.FilterFunction
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

object ProcessFunctionDemo {
  def main(args: Array[String]): Unit = {
    //初始化Flink的Streaming流计算的上下文对象
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度
    env.setParallelism(1)
    //导入隐式转换
    import  org.apache.flink.streaming.api.scala._
    //读取node1通过socket传输过来的数据
    val stream: DataStream[String] = env.socketTextStream("node1",8888)
    /*stream.filter(new FilterFunction[String] {
      override def filter(value: String): Boolean = {
        value.trim.length>0
      }
    }).print()*/
    //stream.filter(_.trim.length>0).print()
    stream.process(new ProcessFunction[String,String](){
      override def processElement(line: String, context: ProcessFunction[String, String]#Context, collector: Collector[String]): Unit = {
        if(line.trim.length>0){
          collector.collect(line)
        }
      }
    }).print()
    env.execute("ProcessFunctionDemo")
  }
}