以Kafka为来源的Flink

Question

我试图从Kafka主题中读取数据，在Flink流媒体。我试图运行以下示例代码，在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)

你能指导我修理这个吗？Kafka连接器是否存在依赖性问题。我的版本是：

1. Flink 1.1.3

Answer 1

因为答案还没有被接受，这里有一个完整的Maven代码示例，可以使用Flink从Kafka读取数据。

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

希望这有帮助。

Answer 2

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

Answer 3

请使用以下版本。它将与您的Kafka版本一起使用。

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

如果它仍然不为你工作，那么请让我知道，我会分享工作代码。