前言

在官方文檔的描述中，API FlinkKafkaConsumer和FlinkKafkaProducer將在后續(xù)版本陸續(xù)棄用、移除，所以在未來生產(chǎn)中有版本升級的情況下，新API KafkaSource和KafkaSink還是有必要學(xué)會使用的。下面介紹下基于新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里的數(shù)據(jù)根據(jù)一定邏輯分發(fā)到不同的目標(biāo)topic里。該類需要繼承 TopicSelector ，這里簡單根據(jù)來源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()
  }
}

自定義序列化器

自定義序列化器可以將數(shù)據(jù)根據(jù)自己的業(yè)務(wù)格式寫到目標(biāo)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()
  }
}

自定義分區(qū)器

自定義分區(qū)器可以根據(jù)具體邏輯對要寫到目標(biāo)Kafka 里的數(shù)據(jù)進(jìn)行partition分配。該類需要繼承 FlinkKafkaPartitioner ，這里根據(jù)key的hash分配到不同的partition里(如果目標(biāo)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環(huán)境
    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
     */
    // 數(shù)據(jù)加工處理，此處省略

    /**
     * sink
     */
? ? // 定義 KafkaSink
    lazy val kafkaSink: KafkaSink[(String, String, String)] =
      KafkaSink.builder[(String, String, String)]()
        // 目標(biāo)集群地址
        .setBootstrapServers("bootstrap.servers")
        // Kafka生產(chǎn)者的各種配置文件，此處省略配置
        .setKafkaProducerConfig(new Properties())
        // 定義消息的序列化模式
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
          // Topic選擇器，使用之前我們自定義的Topic選擇器
          .setTopicSelector(new MyTopicSelector)
          // Key的序列化器，使用之前我們自定義的Key序列化器
          .setKeySerializationSchema(new MyKeySerializationSchema)
          // Value的序列化器，使用之前我們自定義的Value序列化器
          .setValueSerializationSchema(new MyValueSerializationSchema)
          // 自定義分區(qū)器，使用之前我們自定義的自定義分區(qū)器
          .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的簡單使用。大佬們感覺有用的話點個贊吧～??文章來源地址http://www.zghlxwxcb.cn/news/detail-812725.html