Flink以Kafka为来源

Question

我正在尝试从 Flink 流中的 kafka 主题读取数据。我正在尝试运行以下示例代码，该代码作为页面上的示例 APACHE Flink 1.1.3 文档： Apache kafka 连接器，

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();
}

}

我有以下错误：

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)

你能指导我解决这个问题吗？卡夫卡连接器是否存在任何依赖问题。我的版本是：

1. Flink 1.1.3
卡
2. 夫卡 2.10
3. flink-connector-kafka-0.9_2.11-1.0.0.jar

Answer 1

请使用以下版本。它将适用于您的卡夫卡版本。

   <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.1.4</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.9_2.10</artifactId>
            <version>1.1.3</version>
        </dependency>

我在代码中看到编译问题。

更改此设置：

DataStream<String> stream = env
        .addSource(myConsumer)
        .print();

自：

DataStream<String> stream = env
        .addSource(myConsumer);
stream.print();

如果它仍然不适合您，请告诉我，我将分享工作代码。

Answer 2

Flink 和 Flink 连接器的版本必须匹配。将flink-connector依赖项更新为 1.1.3。

Answer 3

由于答案尚未被接受，这里有一个完整的 Maven 代码示例，用于使用 Flink 从 Kafka 读取数据。

您可能需要调整pom.xml以匹配Kafka和Scala版本的设置。

希望这有帮助。