前言
在官方文档的描述中，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.ConsumerRecord
import 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 = false
override 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、value
"TOPIC_" + 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、value
element._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、value
element._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、value
Math.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.OffsetResetStrategy
import 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
*/
// 定义 KafkaSource
lazy 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 source
val inputDS: DataStream[(String, String, String)] = env.fromSource(
kafkaSource,
WatermarkStrategy.noWatermarks(),
"MyKafkaSource")
.setParallelism(1)
/**
* transformation
*/
// 数据加工处理，此处省略
/**
* sink
*/
// 定义 KafkaSink
lazy 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 sink
inputDS.sinkTo(kafkaSink)
.name("MyKafkaSink")
.setParallelism(1)
/**
* execute
*/
env.execute("myJob")
}
}以上就是KafkaSource和KafkaSink API的简单使用。大佬们感觉有用的话点个赞吧～😉
