apache-kafka - Kafka Streams

问题描述

我将用 kafka 流计算平均值。所以我做了一个有状态的操作，聚合，需要创建一个状态存储，但这不会发生。

这里的平均值函数：

private void average () {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<GenericRecord, GenericRecord> source = 
                builder.stream(this.topicSrc);

        KStream <String, Double> average = source
                .mapValues(value -> createJson(value.toString()))
                .map((key, value) -> KeyValue.pair(this.variable, value.getNumberValue(this.pathVariable, this.variable)))
                .groupByKey( Serialized.with(
                        Serdes.String(),
                        Serdes.String()))
                .aggregate (
                        () -> new Tuple(0, 0),
                        (aggKey, newValue, aggValue) ->  new Tuple (aggValue.occ + 1, aggValue.sum + Integer.parseInt(newValue)),
                        Materialized.with(Serdes.String(), new MySerde()))
                .mapValues(v -> v.getAverage())
                .toStream();

        average.to(this.topicDest, Produced.with(Serdes.String(), Serdes.Double()));
        KafkaStreams stream = new KafkaStreams(builder.build(), props);
        stream.start();

    }

这里有一个例外：

 Exception in thread "Thread-0" org.apache.kafka.streams.errors.StreamsException: org.apache.kafka.streams.errors.ProcessorStateException: base state directory [/tmp/kafka-streams] doesn't exist and couldn't be created
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:658)
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:628)
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:538)
    at it.imolinfo.sacmi.processor.Streamer.average(Streamer.java:167)
    at it.imolinfo.sacmi.processor.Streamer.run(Streamer.java:180)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.kafka.streams.errors.ProcessorStateException: base state directory [/tmp/kafka-streams] doesn't exist and couldn't be created
    at org.apache.kafka.streams.processor.internals.StateDirectory.<init>(StateDirectory.java:80)
    at org.apache.kafka.streams.KafkaStreams.<init>(KafkaStreams.java:656)
    ... 5 more

问题是基本目录不存在，但我希望 kafka 流在必要时创建目录。

--- 编辑 ----- 我注意到，如果我有 1 个处理器来计算变量的平均值，则没有问题，但如果我有 2 个处理器，则可以。

1 个处理器的配置文件：

 type->streamer
 number->1
 subtype->average
 variabli->payload:T_DUR_CICLO
 topicSrc->m0-tempi
 topicDest->average
 application.id->stream0
 bootstrap.servers->localhost:9092
 schema.registry.url->http://localhost:8081
 default.key.serde->io.confluent.kafka.streams.serdes.avro.GenericAvroSerde
 default.value.serde->io.confluent.kafka.streams.serdes.avro.GenericAvroSerde

2个处理器的配置文件：

type->streamer
number->1
 subtype->average
 variabli->payload:T_DUR_CICLO
 topicSrc->m0-tempi
 topicDest->average
 application.id->stream0
 bootstrap.servers->localhost:9092
 schema.registry.url->http://localhost:8081
 default.key.serde->io.confluent.kafka.streams.serdes.avro.GenericAvroSerde
 default.value.serde->io.confluent.kafka.streams.serdes.avro.GenericAvroSerde

type->streamer
number->1
 subtype->average
 variabli->payload:HMI_TEMP_E1
 topicSrc->m0-temperature
 topicDest->average
 application.id->stream1
 bootstrap.servers->localhost:9092
 schema.registry.url->http://localhost:8081
 default.key.serde->io.confluent.kafka.streams.serdes.avro.GenericAvroSerde
 default.value.serde->io.confluent.kafka.streams.serdes.avro.GenericAvroSerde

现在我启动处理器：

private void loadStreamer (Tuple t){
    int number = Integer.parseInt(t.getNumber());
    for (int i = 0; i < number; i ++) {
        String[] splitted = t.getVariables()[0].split(":");
        Streamer s = new Streamer (t.getSubType(), t.getTopicSrc(), t.getTopicDest(), splitted[0], splitted[1], t.getProp());
        Thread th = new Thread (s);
        th.start();
    }
}

Tuple 类型包含配置文件的信息。for cicle 中的数字是配置文件中存在的数字。在这种情况下是 1，但我可以为同一个过程做更多的实例来获得污染容差。

流光：

    public class Streamer implements Runnable{


    private final String topicSrc;
    private final String topicDest;
    private final String variable;
    private final String pathVariable;
    private final String type;
    private final Properties props;

 public Streamer (String type, String topicSrc, String topicDest, String pathVariable, String variable, Properties props) {
        this.type = type;
        this.topicSrc = topicSrc;
        this.topicDest = topicDest;
        this.variable = variable;
        this.pathVariable = pathVariable;
        this.props = props;
    }

private void average () {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<GenericRecord, GenericRecord> source = 
                builder.stream(this.topicSrc);

        KStream <String, Double> average = source
                .mapValues(value -> createJson(value.toString()))
                .map((key, value) -> KeyValue.pair(this.variable, value.getNumberValue(this.pathVariable, this.variable) + ":" + value.getStringValue("timestamp")))
                .groupByKey( Serialized.with(
                        Serdes.String(),
                        Serdes.String()))
                .aggregate (
                        () -> new Tuple(0, 0, ""),
                        (aggKey, newValue, aggValue) ->  new Tuple (aggValue.occ + 1, aggValue.sum + Integer.parseInt(newValue.split(":")[0]), newValue.split(":")[1]),
                        Materialized.with(Serdes.String(), new MySerde()))
                .mapValues((key, value) -> new AverageRecord (key, value.getDate(), value.getAverage()))
                .mapValues(v -> v.getAverage())
                .toStream();

        average.to(this.topicDest, Produced.with(Serdes.String(), Serdes.Double()));
        KafkaStreams stream = new KafkaStreams(builder.build(), props);
        stream.start();
    }

 public void run()  {
        switch (this.type) {
            case "average":
                average();
                break;
            case "filter":
                filter();
                break;
            default:
                System.out.println("type not valid " + this.type);
            break;
    }

所以我有 2 个线程和 2 个 Streamer 对象来执行平均功能。与 2 streamer 唯一不同的是计算平均值的变量。

我以错误的方式创建流程？

标签： apache-kafkaapache-kafka-streams

apache-kafka - Kafka Streams - 未创建存储

问题描述

解决方案

推荐阅读