ksql

前提，基于这个项目的dbsinks kafka connector 创建stream create stream stream_name with (kafka_topic='topic', value_format='avro'); SET 'auto.offset.reset' = 'earliest'; run select select after->"ID" from stream_name limit 1;