Flin对接Kafka实践

        
            org.apache.flink
            flink-connector-kafka_2.11
            1.11.2
        

package com.cityos.flinkKafka

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

import java.util.Properties

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

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "host:9092")
    properties.setProperty("group.id", "test3")
    properties.setProperty("auto.offset.reset", "earliest") 

    val stream = env.addSource(new FlinkKafkaConsumer[String]("tracking-test", new SimpleStringSchema(), properties))

    println(stream.parallelism)
    stream.print()


    env.execute(getClass.getCanonicalName)
  }

}

注意 ： (“auto.offset.reset”, “earliest”) 并不是i从头开始消费

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

import java.util.Properties

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

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.readTextFile("data/access_test.log").map(x => x.toString)

    val brokers = "host"
    val topic = "tracking-test"

    val myProducer = new FlinkKafkaProducer[String](
      brokers,
      topic,
      new SimpleStringSchema())

    stream.addSink(myProducer)

    env.execute(getClass.getCanonicalName)
  }
}