gpt4 book ai didi

java - JSON对象: How to map的Kafka流消费者

转载 作者:行者123 更新时间:2023-11-30 01:45:24 27 4
gpt4 key购买 nike

我是 Kafka/Kafka Stream 的新手。我正在使用最新 Kafka/kafka-stream 和 kafka-client 以及 openjdk11。我的生产者正在生成 json 对象(其中 keyname),看起来像

{"Name":"John", "amount":123, "time":2019-10-03T05:24:52" }

生产者代码以便更好地理解:

public static ProducerRecord<String, String> newRandomTransaction(String name) {
// creates an empty json {}
ObjectNode transaction = JsonNodeFactory.instance.objectNode();

Integer amount = ThreadLocalRandom.current().nextInt(0, 100);

// Instant.now() is to get the current time
Instant now = Instant.now();

// we write the data to the json document
transaction.put("name", name);
transaction.put("amount", amount);
transaction.put("time", now.toString());
return new ProducerRecord<>("bank-transactions", name, transaction.toString());
}

现在我正在尝试编写消耗交易并计算该人余额中的总金额的应用程序。

(仅供引用:我正在使用旧代码并尝试使其工作)。

使用GroupBYKey,因为主题已经有了正确的键。然后聚合来计算我正在努力的总余额。

此刻的应用程序(注释掉的部分是我试图使其在下一行中工作的旧代码):

public class BankBalanceExactlyOnceApp {
private static ObjectMapper mapper = new ObjectMapper();

public static void main(String[] args) {
Properties config = new Properties();

config.put(StreamsConfig.APPLICATION_ID_CONFIG, "bank-balance-application");
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

// we disable the cache to demonstrate all the "steps" involved in the transformation - not recommended in prod
config.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, "0");

// Exactly once processing!!
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE);

// json Serde
final Serializer<JsonNode> jsonSerializer = new JsonSerializer();
final Deserializer<JsonNode> jsonDeserializer = new JsonDeserializer();
final Serde<JsonNode> jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);


StreamsBuilder builder = new StreamsBuilder();

KStream<String, JsonNode> bankTransactions =
builder.stream( "bank-transactions", Materialized.with(Serdes.String(), jsonSerde);


// create the initial json object for balances
ObjectNode initialBalance = JsonNodeFactory.instance.objectNode();
initialBalance.put("count", 0);
initialBalance.put("balance", 0);
initialBalance.put("time", Instant.ofEpochMilli(0L).toString());

/*KTable<String, JsonNode> bankBalance = bankTransactions
.groupByKey(Serdes.String(), jsonSerde)
.aggregate(
() -> initialBalance,
(key, transaction, balance) -> newBalance(transaction, balance),
jsonSerde,
"bank-balance-agg"
);*/

KTable<String, JsonNode> bankBalance = bankTransactions
.groupByKey(Serialized.with(Serdes.String(), jsonSerde))
.aggregate(
() -> initialBalance,
(key, transaction, balance) -> {
//String t = transaction.toString();
newBalance(transaction, balance);
},
Materialized.with(Serdes.String(), jsonSerde),
"bank-balance-agg"
);

bankBalance.toStream().to("bank-balance-exactly-once", Produced.with(Serdes.String(), jsonSerde));

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.cleanUp();
streams.start();

// print the topology
System.out.println(streams.toString());

// shutdown hook to correctly close the streams application
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}

private static JsonNode newBalance(JsonNode transaction, JsonNode balance) {
// create a new balance json object
ObjectNode newBalance = JsonNodeFactory.instance.objectNode();
newBalance.put("count", balance.get("count").asInt() + 1);
newBalance.put("balance", balance.get("balance").asInt() + transaction.get("amount").asInt());

Long balanceEpoch = Instant.parse(balance.get("time").asText()).toEpochMilli();
Long transactionEpoch = Instant.parse(transaction.get("time").asText()).toEpochMilli();
Instant newBalanceInstant = Instant.ofEpochMilli(Math.max(balanceEpoch, transactionEpoch));
newBalance.put("time", newBalanceInstant.toString());
return newBalance;
}
}

问题是:当我尝试在行中调用newBalance(transaction,balance)时:

aggregate(
() -> initialBalance,
(key, transaction, balance) -> newBalance(transaction, balance),
jsonSerde,
"bank-balance-agg"
)

并使用 msg 查看编译器错误:

newBalance(JsonNode, JsonNode) can not be applied to (<lambda parameter>,<lambda parameter>)

我尝试将其读取为字符串,将参数类型从 JsonNode 更改为 Object。但是,无法修复它。

我可以得到有关如何修复它的任何建议吗?

最佳答案

Kafka Streams 2.3 中的

KGroupedStream 没有具有以下签名的方法:

<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
final Aggregator<? super K, ? super V, VR> aggregator,
final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized,
String aggregateName);

有两个重载方法aggregate:

<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
final Aggregator<? super K, ? super V, VR> aggregator);

<VR> KTable<K, VR> aggregate(final Initializer<VR> initializer,
final Aggregator<? super K, ? super V, VR> aggregator,
final Materialized<K, VR, KeyValueStore<Bytes, byte[]>> materialized);

您应该使用第二个,您的代码应类似于:

KTable<String, JsonNode> bankBalance = input
.groupByKey(Grouped.with(Serdes.String(), jsonSerde))
.aggregate(
() -> initialBalance,
(key, transaction, balance) -> newBalance(transaction, balance),
Materialized.with(Serdes.String(), jsonSerde)
);

关于java - JSON对象: How to map的Kafka流消费者,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/58225287/

27 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com