读写Kafka比较简单，官方提供了connector，也提供了例子可以参看，官网例子的GitHub地址：

https://github.com/apache/flink/tree/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/kafka

下面只做简单的说明：

    val myConsumer = new FlinkKafkaConsumer010[String](KAFKA_CONSUMER_TOPIC, new SimpleStringSchema(), KAFKA_PROP)//参数分别是：指定取数据的topic，序列化器，kafka参数
    env.addSource[String](myConsumer)//读入kafka数据产生dataStream，可针对dataStream进行各种逻辑处理，比如map，filter等等。

//处理结果写入kafka
 dataStream.addSink(new FlinkKafkaProducer010[String](KAFKA_PRODUCER_TOPIC,new SimpleStringSchema(),KAFKA_PROP))//参数分别是：写入topic，序列化器，kafka配置惨

//kafka配置参数
  lazy val KAFKA_PROP: Properties = new Properties() {
    setProperty("bootstrap.servers", KAFKA_BROKER)//broker地址
    setProperty("zookeeper.connect", KAFKA_ZOOKEEPER_HOST)//zookeeper配置
    setProperty("group.id", KAFKA_GROUP_ID)//组id
  }