object Keyby {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(3)

    val textStream: DataStream[String] = env.socketTextStream("localhost" , 12345)

    val flatMap_data: DataStream[String] = textStream.flatMap(line => line.split("\t"))

    val map_data: DataStream[(String, Int)] = flatMap_data.map(line => (line , 1))

    //TODO 逻辑上将一个流分成不相交的分区，每个分区包含相同键的元素。在内部，这是通过散列分区来实现的

    val keyByData: KeyedStream[(String, Int), String] = map_data.keyBy(line => line._1)

    keyByData.writeAsText("keyByData")

    env.execute()

  }

}