2023学年完整公开课版Flink 实时APIDataSink

Flink实时API-DataSinkDataStream——SinkDataStreamAPISourceTransformationsSinkDataStream——SinkDataStreamAPISourceTransformationsSink数据接收器Sink将DataStream中的数转发到文件，套接字，外部系统或打印它们。1）writeAsText()：-将元素按行写为字符串。

通过调用每个元素的toString（）方法获得字符串。2）writeToSocket：根据一个元素将元素写入套接字3）Print()：打印每个元素值到标准输出流中4）addSink：调用自定义接受器功能DataStream——Sink1）writeAsText()：将元素按行写为字符串。通过调用每个元素的toString（）方法获得字符串。2）writeToSocket：根据一个元素将元素写入套接字DataStream——Sink

//换一个接受的节点env.socketTextStream("localhost",port=8989).writeToSocket("localhost",port=8989,newSimpleStringSchema())env.execute()valenv=StreamExecutionEnvironment.getExecutionEnvironment//多并行度env.socketTextStream("localhost",port=8989).writeAsText("D:\\project\\first_flink01\\src\\main\\resources")Connectors3）addSink：调用自定义接收器

Connectors即连接器，实现了某个组件的Source或Sink，我们可以使用官方内置Connectors实现对数据流的实时读取或实时写出。官方内置Connectors有如下：ApacheKafka(source/sink)ApacheCassandra(sink)Elasticsearch(sink)HadoopFileSystem(sink)RabbitMQ(source/sink)ApacheActiveMQ(source/sink)Redis(sink)Kafka

Source&Sink

Flink生产者算子转换操作主题：t1KafkaSource消费者(主题：t1）KafkaSink生产者(主题：t2）消费者主题：t2算子转换操作KafkaSource消费者(主题：t1）Kafka

Source&Sinkflink既可以从kafka中消费数据，也能将数据写入kafka。官方已经内置好Kafka的Connector。使用时需要导入maven依赖读取步骤：①获取kafkaSource②调用env.addSource(kafkaSource)写入步骤：①获取kafkaSink②调用env.addSink(kafkaSink)Kafka

Source&Sinkflink既可以从kafka中消费数据，也能将数据写入kafka。官方已经内置好Kafka的Connector。使用时需要导入maven依赖读取步骤：①获取kafkaSource②调用env.addSource(kafkaSource)写入步骤：①获取kafkaSink②调用env.addSink(kafkaSink)Kafka

Source&SinkobjectKafkaUtil{valprop=newProperties()valtopic="t1"//设置kafka服务器地址

prop.setProperty("bootstrap.servers","ubuntu:9092")//设置kafka消费者

prop.setProperty("group.id","con1")//获得KafkaSource，类似于消费者

defgetKafkaSource(topic:String):FlinkKafkaConsumer09[String]={newFlinkKafkaConsumer09[String](topic,newSimpleStringSchema(),prop)}

//获得KafkaSink,此sink负责将数据写入到Kafka，类似于生产者

defgetKafkaSink(topic:String):FlinkKafkaProducer09[String]={newFlinkKafkaProducer09[String](prop.getProperty("bootstrap.servers"),topic,newSimpleStringSchema())

}}Kafka

Source&SinkobjectKafkaSourceAndSinkDemo{defmain(args:Array[String]):Unit={valenv=StreamExecutionEnvironment.getExecutionEnvironment

//获取KafkaSourcevalkafkaSource=KafkaUtil.getKafkaSource("event_log")//获取KafkaSinkvalkafkaSink=