文报 2020-01-09
方式一:读取文件输出到Kafka
1.代码
import org.apache.flink.api.common.serialization.SimpleStringSchemaimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011//温度传感器读取样例类case class SensorReading(id: String, timestamp: Long, temperature: Double)object KafkaSinkTest { def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) import org.apache.flink.api.scala._ val inputStream = env.readTextFile("sensor.txt") val dataStream = inputStream.map(x => { val arr = x.split(",") SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble).toString //转成String方便序列化输出 }) //sink dataStream.addSink(new FlinkKafkaProducer011[String]("localhost:9092", "sinkTest", new SimpleStringSchema())) dataStream.print() env.execute(" kafka sink test") }}2.启动zookeeper:参考https://www.cnblogs.com/wddqy/p/12156527.html3.启动kafka:参考https://www.cnblogs.com/wddqy/p/12156527.html4.创建kafka消费者观察结果
方式二:Kafka到Kafka
1.代码
import java.util.Propertiesimport org.apache.flink.api.common.serialization.SimpleStringSchemaimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}//温度传感器读取样例类case class SensorReading(id: String, timestamp: Long, temperature: Double)object KafkaSinkTest1 { def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) import org.apache.flink.api.scala._ //从Kafka到Kafka val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") properties.setProperty("group.id", "consumer-group") properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") properties.setProperty("auto.offset.reset", "latest") val inputStream = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties)) val dataStream = inputStream.map(x => { val arr = x.split(",") SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble).toString //转成String方便序列化输出 }) //sink dataStream.addSink(new FlinkKafkaProducer011[String]("localhost:9092", "sinkTest", new SimpleStringSchema())) dataStream.print() env.execute(" kafka sink test") }}
2.启动zookeeper:参考https://www.cnblogs.com/wddqy/p/12156527.html3.启动kafka:参考https://www.cnblogs.com/wddqy/p/12156527.html4.创建Kafka生产者和消费者,运行代码,观察结果