flink kafka source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

object FlinkDemo05_KafkaSource {
    
    val prop = new Properties()
    prop.setProperty("bootstrap.servers", "linux01:9092")
    prop.setProperty("group.id", "flink-grp")
    prop.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty("auto.offset.reset", "latest")
    
    def main(args: Array[String]): Unit = {
        //1 创建环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //2 获取Stream
        import org.apache.flink.api.scala._
        val topic = "flink-topic"
        val schema = new SimpleStringSchema()
        val dStream: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String](topic, schema, prop))
        //3 计算
        val result = dStream.flatMap(_.split("\s")).filter(_.nonEmpty).map((_,1)).keyBy(0).sum(1)
        result.print()
        
        //4 执行
        env.execute("kafka source job")
        
    }
}