我有两个Kafka的钥匙 String
价值观 Avro
我用ksql创建的格式。
第一个是:
DESCRIBE EXTENDED STREAM_1;
Type : STREAM
Key field : IDUSER
Timestamp field : Not set - using <ROWTIME>
Key format : STRING
Value format : AVRO
Kafka output topic : STREAM_1 (partitions: 4, replication: 1)
Field | Type
--------------------------------------------------------
ROWTIME | BIGINT (system)
ROWKEY | VARCHAR(STRING) (system)
FIRSTNAME | VARCHAR(STRING)
LASTNAME | VARCHAR(STRING)
IDUSER | VARCHAR(STRING)
第二个是:
DESCRIBE EXTENDED STREAM_2;
Type : STREAM
Key field : IDUSER
Timestamp field : Not set - using <ROWTIME>
Key format : STRING
Value format : AVRO
Kafka output topic : STREAM_2 (partitions: 4, replication: 1)
Field | Type
--------------------------------------------------------
ROWTIME | BIGINT (system)
ROWKEY | VARCHAR(STRING) (system)
USERNAME | VARCHAR(STRING)
IDUSER | VARCHAR(STRING)
DEVICE | VARCHAR(STRING)
所需输出应包括 IDUSER
, LASTNAME
, DEVICE
以及 USERNAME
.
我想 left join
这些溪流 IDUSER
)使用streams api并将输出写入kafka主题。
为此,我尝试了以下方法:
public static void main(String[] args) {
final Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-strteams");
streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
streamsConfiguration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
final Serde<String> stringSerde = Serdes.String();
final Serde<GenericRecord> genericAvroSerde = new GenericAvroSerde();
boolean isKeySerde = false;
genericAvroSerde.configure(Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"), isKeySerde);
KStreamBuilder builder = new KStreamBuilder();
KStream<String, GenericRecord> left = builder.stream("STREAM_1");
KStream<String, GenericRecord> right = builder.stram("STREAM_2");
// Java 8+ example, using lambda expressions
KStream<String, GenericRecord> joined = left.leftJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
JoinWindows.of(TimeUnit.MINUTES.toMillis(5)),
Joined.with(
stringSerde, /* key */
genericAvroSerde, /* left value */
genericAvroSerde) /* right value */
);
joined.to(stringSerde, genericAvroSerde, "streams-output-testing");
KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
streams.cleanUp();
streams.start();
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
然而,
KStream<String, GenericRecord> joined = ...
在我的ide上引发错误:
incompatible types: inference variable VR has incompatible bounds
当我尝试使用
String Serde
对于键和值,它都起作用,但数据不是从中可读的 kafka-console-consumer
. 我想做的是以avro格式生成数据,以便能够使用 kafka-avro-console-consumer
.
2条答案
按热度按时间k7fdbhmy1#
我的第一个猜测是你正在返回一个
String
而您的代码需要GenericRecord
结果是:注意如何
joined
有类型KStream<String, GenericRecord>
,即值具有类型GenericRecord
,但联接输出是通过"left=" + leftValue + ", right=" + rightValue
,具有类型String
.zfycwa2u2#
您可以直接返回值,而不是将值转换为字符串。例如: