使用 FLINK 将 JSON 下沉到 Kafka 的最快方法

Question

代码优化

我有一个从 url/port 读取数据的 flink 应用程序对其执行处理并返回 JSON。然后我将 JSON 转换为 String 并将其接收到 Kafka。

当前性能和注意问题

如果我只是执行处理 -> 我可以通过该函数运行大约 30,000 个字符串，但是当我添加函数以将其转换为 STring 然后下沉到 kafka 时，我的吞吐量下降到每秒 17,000 个字符串。

在下沉到 Kafka 之前是否需要将我的 JSON 转换为字符串？如果不是，我如何将 json ObjectNode 下沉到 kafka？

还有什么其他解决方案。 我认为瓶颈是字符串函数

我尝试使用几种方法（.toString 函数、StringBuilder 到 String）将 JSON 转换为字符串。

 // Read from Source
 val in_stream = env.socketTextStream(url, port,      socket_stream_deliminator, socket_connection_retries).setParallelism(1)

 // Perform Process
 .map(x=>{Process(x)}).setParallelism(1)

 // Convert to STring
 .map(x => ObjectNodeToString({
     val json_string_builder = StringBuilder.newBuilder
     json_string_builder.append(x)
     return json_string_builder.toString()
 })).setParallelism(1)

 // sink data
 .addSink(new FlinkKafkaProducer[String](broker_hosts, global_topic, new SimpleStringSchema()))

我想保持每秒处理 30,000 个字符串。我没有转换为字符串函数。我可以将 ObjectNode 直接下沉到 kafka 吗？

Answer 1

你可以。 Sink 在将给定对象发送到 kafka 之前将其序列化为字节数组。确保您的接收器函数提供了能够将 ObjectNode 转换为字节数组的序列化程序。

还要确保消费者准备好接收 ObjectNode 对象，而不是字符串。