FlinkKafkaProducer的重载构造函数

Question

我正在尝试使用Scala和Flink将消息发布到Kafka主题中。但是，当使用the documentation中提供的代码创建FlinkKafkaProducer对象时，它告诉我无法应用构造器。以下是代码示例：

val studentProducer = new FlinkKafkaProducer[String](
 "my_topic",                  // target topic
 new SimpleStringSchema(),    // serialization schema
 properties,                  // producer config
 FlinkKafkaProducer.Semantic.EXACTLY_ONCE) // fault-tolerance

具有以下导入：

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}
import org.apache.flink.streaming.util.serialization.SimpleStringSchema

import java.util.Properties

这就是我得到的错误：

/home/user/Flink/flinkproj/src/main/scala/org/flink/Job.scala:83:27: overloaded method constructor FlinkKafkaProducer with alternatives:
[error]   (x$1: String,x$2: org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema[String],x$3: java.util.Properties,x$4: org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.Semantic)org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer[String] <and>
[error]   (x$1: String,x$2: org.apache.flink.streaming.util.serialization.KeyedSerializationSchema[String],x$3: java.util.Properties,x$4: java.util.Optional[org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner[String]])org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer[String] <and>
[error]   (x$1: String,x$2: org.apache.flink.streaming.util.serialization.KeyedSerializationSchema[String],x$3: java.util.Properties,x$4: org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.Semantic)org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer[String] <and>
[error]   (x$1: String,x$2: org.apache.flink.api.common.serialization.SerializationSchema[String],x$3: java.util.Properties,x$4: java.util.Optional[org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner[String]])org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer[String]
[error]  cannot be applied to (String, org.apache.flink.streaming.util.serialization.SimpleStringSchema, java.util.Properties, org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.Semantic)
[error]     val studentProducer = new FlinkKafkaProducer[String](

变量properties是java.util.Properties的一个实例。我认为它必须在字符串序列化程序中，但我看不出有什么问题。

以下是来自build.sbt的版本的详细信息

ThisBuild / scalaVersion := "2.11.8"

val flinkVersion = "1.11.3"

val flinkDependencies = Seq(
  "org.apache.flink" %% "flink-clients" % flinkVersion % "provided",
  "org.apache.flink" %% "flink-scala" % flinkVersion % "provided",
  "org.apache.flink" %% "flink-streaming-scala" % flinkVersion % "provided",
  "org.apache.flink" %% "flink-connector-kafka" % flinkVersion % "provided")

Answer 1

我认为文档已经过时了，您需要提供KafkaSerializationSchema或KeyedSerializationSchema --或者如果您使用SerializationSchema，则还需要提供FlinkKafkaPartitioner。

我没有用Scala写的例子，但是这里有一个用Java写的例子，展示了如何实现一个使用ObjectMapper写出JSON的KafkaSerializationSchema：

/**
 * A Kafka {@link KafkaSerializationSchema} to serialize {@link ClickEventStatistics}s as JSON.
 *
 */
public class ClickEventStatisticsSerializationSchema implements KafkaSerializationSchema<ClickEventStatistics> {

    private static final ObjectMapper objectMapper = new ObjectMapper();
    private String topic;

    public ClickEventStatisticsSerializationSchema(){
    }

    public ClickEventStatisticsSerializationSchema(String topic) {
        this.topic = topic;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(
            final ClickEventStatistics message, @Nullable final Long timestamp) {
        try {
            //if topic is null, default topic will be used
            return new ProducerRecord<>(topic, objectMapper.writeValueAsBytes(message));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Could not serialize record: " + message, e);
        }
    }
}