时间戳

首页 > 解决方案 > 如何将 Protobuf 数据从 Flink 转发到 Kafka 和 stdout？

scala - 如何将 Protobuf 数据从 Flink 转发到 Kafka 和 stdout？

问题描述

我想在这里添加一些代码，并标准输出来自 Flink 的 protobuf 数据。

我正在使用 Flink 的 Apache Kafka 连接器将 Flink 连接到 Kafka。

这是我的 Flink 的代码。

val env = StreamExecutionEnvironment.getExecutionEnvironment
val props = new Properties()
props.setProperty("bootstrap.servers", "localhost:9092").
val producer = FlinkKafkaProducer011(topic, new myProtobufSchema, props)
env.addSink(producer)
env.execute("To Kafka")

这是我的卡夫卡的代码。

  val props: Properties = {
    val p = new Properties()
    p.put(StreamsConfig.APPLICATION_ID_CONFIG, "protobuf-application")
    p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    p
  }

  val builder: StreamsBuilder = new StreamsBuilder
  // TODO: implement here to stdout 

  val streams: KafkaStreams = new KafkaStreams(builder.build(), props)
  streams.start()

  sys.ShutdownHookThread {
     streams.close(Duration.ofSeconds(10))
  }

标签： scalaapache-kafkaapache-flinkapache-kafka-streams

解决方案

您需要设置StreamsBuilder从主题消费

val builder: StreamsBuilder = new StreamsBuilder()
    .stream(topic)
    .print(Printed.toSysOut());

推荐阅读