Ich versuche, die Daten von Kafka Thema in Flink Streaming zu lesen. Ich versuche, den folgenden Beispiel-Code auszuführen, die dort auf Seite APACHE Flink 1.1.3 Dokumentation als ein Beispiel ist: Apache kafka-Anschluss,Flink mit Kafka als Quelle
import java.util.Properties;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.flink.streaming.util.serialization.SerializationSchema;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
public class stock_streaming_kafka {
public static void main(String[] args) throws Exception
{
StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
FlinkKafkaConsumer09<String> myConsumer = new FlinkKafkaConsumer09<>("nsestocks4k", new SimpleStringSchema(), properties);
DataStream<String> stream = env
.addSource(myConsumer)
.print();
}
}
Ich habe folgende Fehlermeldung:
Exception in thread "main" java.lang.Error: Unresolved compilation problems:
The type org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase cannot be resolved. It is indirectly referenced from required .class files
The method addSource(SourceFunction<OUT>) in the type StreamExecutionEnvironment is not applicable for the arguments (FlinkKafkaConsumer09<String>)
at stock_streaming_kafka.main(stock_streaming_kafka.java:25)
Können Sie mir bitte helfen, dies zu beheben? Gibt es ein Abhängigkeitsproblem mit dem Kafka-Connector? Meine Versionen sind:
- Flink 1.1.3
- Kafka 2.10
- flink-connector-kafka-0.9_2.11-1.0.0.jar
Danke Fabian. Ich habe die jar-Version in flink-connector-kafka-0.9_2.11-1.1.3.jar geändert. Aber ich habe den Fehler als _Typ org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema kann nicht aufgelöst werden. Es wird indirekt von erforderlichen .class files_ referenziert – kadsank