flink_初识02kafkawordcount

package flink

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.flink.streaming.api.scala._

object KafkaWordCount {


  def getFlinkKafkaConsumer() = {
    val prop = new Properties()

    prop.setProperty("zookeeper.connect", "localhost:2181") //ZOOKEEPER_HOST
    prop.setProperty("bootstrap.servers", "localhost:9092") //KAFKA_BROKER
    prop.setProperty("group.id", "group1") //TRANSACTION_GROUP
    new FlinkKafkaConsumer011[String]("test_flink", new SimpleStringSchema(), prop) //TOPIC
  }


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

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val source = getFlinkKafkaConsumer()

    source.setStartFromEarliest()

    val dStream = env.addSource(source)

    val result = dStream.flatMap(x => x.split("\\s"))
      .map(x => (x, 1)).keyBy(0).timeWindow(Time.seconds(2l)).sum(1)

    result.setParallelism(1).print()

    result.writeAsText("E:\\\\sparkproject\\\\src\\\\test\\\\data\\\\result2.txt", WriteMode.OVERWRITE)

    env.execute("KafkaWordCount")
  }

}