Когда я пытаюсь написать свое собственное демо Apache Beam, используя Kafka для публикации / подсчета данных из системы покупок, используя Beam для проектирования потока данных, работайте на Flink.Я застрял в очень редком исключении:
Caused by: java.lang.IncompatibleClassChangeError: Found interface org.apache.flink.streaming.api.operators.InternalTimer, but class was expected
at org.apache.beam.runners.flink.translation.wrappers.streaming.WindowDoFnOperator.fireTimer(WindowDoFnOperator.java:129)
at org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator.onProcessingTime(DoFnOperator.java:704)
at org.apache.flink.streaming.api.operators.InternalTimerServiceImpl.onProcessingTime(InternalTimerServiceImpl.java:235)
at org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$TriggerTask.run(SystemProcessingTimeService.java:285)
мой код:
package com.meikeland.dataflow;
import org.apache.beam.runners.flink.FlinkRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.*;
import org.apache.beam.sdk.transforms.windowing.*;
import org.apache.beam.sdk.values.KV;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class GameStats {
private static final Logger logger = LoggerFactory.getLogger(GameStats.class);
public static void main(String[] args) {
KFOptions options = PipelineOptionsFactory.fromArgs(args).as(KFOptions.class);
options.setRunner(FlinkRunner.class);
options.setStreaming(true);
logger.info("brokers address is: {}", options.getBrokers());
runDemoCount(options);
}
private static void runDemoCount(KFOptions options) {
Pipeline pipeline = Pipeline.create(options);
pipeline
// read order events from kafka
.apply("ConsumeKafka",
KafkaIO.<Long, String>read().withBootstrapServers(options.getBrokers()).withTopic("tracking.order.goods")
.withKeyDeserializer(LongDeserializer.class).withValueDeserializer(StringDeserializer.class)
.withLogAppendTime().withoutMetadata())
.apply(Values.create()).apply("ParseOrderInfo", ParDo.of(new ParseOrderInfoFn()))
.apply("SetTimestamp", WithTimestamps.of(OrderInfo::getCreatedAt))
.apply("ExtractOrderID", MapElements.via(new SimpleFunction<OrderInfo, Integer>() {
public Integer apply(OrderInfo o) {
logger.info("processed orderID: {}", o.getOrderID());
return o.getOrderID();
}
}))
// window
.apply("FixedWindowsOrderID",
Window.<Integer>into(FixedWindows.of(new Duration(1000 * 60)))
.triggering(AfterWatermark.pastEndOfWindow()
.withEarlyFirings(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(new Duration(1000 * 60)))
.withLateFirings(AfterPane.elementCountAtLeast(1)))
.withAllowedLateness(new Duration(1000 * 60)).accumulatingFiredPanes())
.apply("Count", Count.<Integer>perElement()).apply("ToString", ParDo.of(new DoFn<KV<Integer, Long>, String>() {
@ProcessElement
public void processElement(@Element KV<Integer, Long> element, IntervalWindow window,
OutputReceiver<String> r) {
logger.info("the order is : {}, and count is : {}", element.getKey(), element.getValue());
r.output(String.format("interval :%s, Order ID: %d, Count :%d", window.start().toString(), element.getKey(),
element.getValue()));
}
})).apply("WriteToKafka", KafkaIO.<Void, String>write().withBootstrapServers(options.getBrokers())
.withTopic("streaming.order.count").withValueSerializer(StringSerializer.class).values());
pipeline.run().waitUntilFinish();
}
}
Кажется, ошибка в окне, но я не могупонять это.И гуглю везде, похоже никто не сталкивался с подобной ошибкой.Поэтому я, должно быть, ошибаюсь.Пожалуйста, кто может спасти меня.