张家界网站,微信平台的微网站怎么做,网站首页被降权怎么做,注册城乡规划师考试题型前言 在官方文档的描述中#xff0c;API FlinkKafkaConsumer和FlinkKafkaProducer将在后续版本陆续弃用、移除#xff0c;所以在未来生产中有版本升级的情况下#xff0c;新API KafkaSource和KafkaSink还是有必要学会使用的。下面介绍下基于新API的一些自定义类以及主程序的… 前言 在官方文档的描述中API FlinkKafkaConsumer和FlinkKafkaProducer将在后续版本陆续弃用、移除所以在未来生产中有版本升级的情况下新API KafkaSource和KafkaSink还是有必要学会使用的。下面介绍下基于新API的一些自定义类以及主程序的简单实践。 官方案例 官方文档地址: https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/connectors/datastream/kafka/ KafkaSource的自定义类 自定义反序列化器 自定义反序列化器可以以指定的格式取到来源Kafka消息中我们想要的元素。该类需要继承 KafkaDeserializationSchema 这里简单将来源Kafka的topic、key、value以Tuple3[String, String, String]的格式取出来。 MyKafkaDeserializationSchemaTuple3.scala import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema
import org.apache.kafka.clients.consumer.ConsumerRecordimport java.nio.charset.StandardCharsets/*** author hushhhh*/
class MyKafkaDeserializationSchemaTuple3 extends KafkaDeserializationSchema[(String, String, String)] {override def deserialize(record: ConsumerRecord[Array[Byte], Array[Byte]]): (String, String, String) {new Tuple3[String, String, String](record.topic(),new String(record.key(), StandardCharsets.UTF_8),new String(record.value(), StandardCharsets.UTF_8))}override def isEndOfStream(nextElement: (String, String, String)): Boolean falseoverride def getProducedType: TypeInformation[(String, String, String)] {TypeInformation.of(classOf[(String, String, String)])}
} KafkaSink的自定义类 自定义Topic选择器 自定义一个 TopicSelector 可以将流中多个topic里的数据根据一定逻辑分发到不同的目标topic里。该类需要继承 TopicSelector 这里简单根据来源Kafka的topic名拼接下。 MyTopicSelector.scala import org.apache.flink.connector.kafka.sink.TopicSelector/*** author hushhhh*/
class MyTopicSelector extends TopicSelector[(String, String, String)] {override def apply(t: (String, String, String)): String {// t: 来源kafka的topic、key、valueTOPIC_ t._1.toUpperCase()}
} 自定义序列化器 自定义序列化器可以将数据根据自己的业务格式写到目标Kafka的key和value里这里将来源Kafka里的key和value直接写出去这两个类都需要继承 SerializationSchema 。 ProducerRecord Key的序列化器 MyKeySerializationSchema.scala import org.apache.flink.api.common.serialization.SerializationSchema/*** author hushhhh*/
class MyKeySerializationSchema extends SerializationSchema[(String, String, String)] {override def serialize(element: (String, String, String)): Array[Byte] {// element: 来源kafka的topic、key、valueelement._2.getBytes()}
} ProducerRecord Value的序列化器 MyValueSerializationSchema.scala import org.apache.flink.api.common.serialization.SerializationSchema/*** author hushhhh*/
class MyValueSerializationSchema extends SerializationSchema[(String, String, String)] {override def serialize(element: (String, String, String)): Array[Byte] {// element: 来源kafka的topic、key、valueelement._3.getBytes()}
} 自定义分区器 自定义分区器可以根据具体逻辑对要写到目标Kafka 里的数据进行partition分配。该类需要继承 FlinkKafkaPartitioner 这里根据key的hash分配到不同的partition里(如果目标topic有多个partition的话)。 MyPartitioner.scala import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner/*** author hushhhh*/
class MyPartitioner extends FlinkKafkaPartitioner[(String, String, String)] {override def partition(record: (String, String, String), key: Array[Byte], value: Array[Byte], targetTopic: String, partitions: Array[Int]): Int {// record: 来源kafka的topic、key、valueMath.abs(new String(record._2).hashCode % partitions.length)}
} 主类 Main.scala import format.{MyKafkaDeserializationSchemaTuple3, MyKeySerializationSchema, MyPartitioner, MyTopicSelector, MyValueSerializationSchema}
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.connector.base.DeliveryGuarantee
import org.apache.flink.connector.kafka.sink.{KafkaRecordSerializationSchema, KafkaSink}
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema
import org.apache.kafka.clients.consumer.OffsetResetStrategyimport java.util.Properties
import scala.collection.JavaConverters._/*** author hushhhh*/
object Main {def main(args: Array[String]): Unit {/*** env*/// stream环境val env: StreamExecutionEnvironment StreamExecutionEnvironment.getExecutionEnvironment/*** source*/// 定义 KafkaSourcelazy val kafkaSource: KafkaSource[(String, String, String)] KafkaSource.builder()// Kafka消费者的各种配置文件此处省略配置.setProperties(new Properties())// 配置消费的一个或多个topic.setTopics(sourceTopic1,sourceTopic2,....split(,, -1).toList.asJava)// 开始消费位置从已提交的offset开始消费没有的话从最新的消息开始消费.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))// 反序列化使用之前我们自定义的反序列化器.setDeserializer(KafkaRecordDeserializationSchema.of(new MyKafkaDeserializationSchemaTuple3)).build()// 添加 kafka sourceval inputDS: DataStream[(String, String, String)] env.fromSource(kafkaSource,WatermarkStrategy.noWatermarks(),MyKafkaSource).setParallelism(1)/*** transformation*/// 数据加工处理此处省略/*** sink*/// 定义 KafkaSinklazy val kafkaSink: KafkaSink[(String, String, String)] KafkaSink.builder[(String, String, String)]()// 目标集群地址.setBootstrapServers(bootstrap.servers)// Kafka生产者的各种配置文件此处省略配置.setKafkaProducerConfig(new Properties())// 定义消息的序列化模式.setRecordSerializer(KafkaRecordSerializationSchema.builder()// Topic选择器使用之前我们自定义的Topic选择器.setTopicSelector(new MyTopicSelector)// Key的序列化器使用之前我们自定义的Key序列化器.setKeySerializationSchema(new MyKeySerializationSchema)// Value的序列化器使用之前我们自定义的Value序列化器.setValueSerializationSchema(new MyValueSerializationSchema)// 自定义分区器使用之前我们自定义的自定义分区器.setPartitioner(new MyPartitioner).build())// 语义保证保证至少一次.setDeliverGuarantee(DeliveryGuarantee.AT_LEAST_ONCE).build()// 添加 kafka sinkinputDS.sinkTo(kafkaSink).name(MyKafkaSink).setParallelism(1)/*** execute*/env.execute(myJob)}} 以上就是KafkaSource和KafkaSink API的简单使用。大佬们感觉有用的话点个赞吧
