Flink 批处理算子详解

 val env = ExecutionEnvironment.getExecutionEnvironment

val text = env.fromElements("who's there","I think I hear")

    text.flatMap{_.toLowerCase.split("\\w+") filter(_.nonEmpty)}
        .map((_,1))
        .groupBy(0)
        .sum(1)

counts.print();

env.execute("Kafka Dataset WordCount")

    val parameter = new Configuration
    parameter.setBoolean("recursive.file.enumeration",true)
    env.readTextFile("file://path/with/files").withParameters(parameter)

    val input: DataSet[(Int,String,Double)] = env.fromElements(
      (1,"hello",4),
      (1,"hello",5),
      (2,"hello",5),
      (3,"word",6),
      (3,"word",6)
    )
    val value = input.groupBy(1).aggregate(Aggregations.SUM,0).aggregate(Aggregations.MIN,2)

    val input1 = env.fromElements((1,"hello"),(2,"hello"))
    val input2 = env.fromElements(("hello",1),("word",2))
    
    val result = input1.join(input2).where(0).equalTo(1)

val toBroadcast = env.fromElements(1,2,3);

val toBroadcast = env.fromElements(1,2,3);
  val toBroadcast:DataSet[Int] = env.fromElements(1,2,3);

    toBroadcast.map(new RichMapFunction[String,String]() {
      var broadcastSet  = null

      override def open(parameters: Configuration): Unit = {
        // 读取广播变量
        broadcastSet = getRuntimeContext.getBroadcastVariable[String]("broadcastSetName").get(0)
      }
      override def map(value: String): String = {
        
      }
      // 注册广播变量
    }).withBroadcastSet(toBroadcast,"broadcastSetName");