0

我从 Apache Flink 的动手示例中复制了示例代码，并尝试让它运行。代码如下：

public class RideCleansing {

    private static final String LOCAL_KAFKA_BROKER = "localhost:9092";
    public static final String CLEANSED_RIDES_TOPIC = "mytopic";


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<TaxiRide> rides = env.addSource(new TaxiRideGenerator("C://data/nycTaxiRides.gz", 1000.0f));

        DataStream<TaxiRide> filteredRides = rides.filter(new NYCFilter());

        filteredRides.addSink(new FlinkKafkaProducer<>(LOCAL_KAFKA_BROKER,
                CLEANSED_RIDES_TOPIC,
                new TaxiRideSchema()));

        env.execute("Taxi Ride Cleansing");
    }

我在本地安装并运行了 Apache Kafka。我收到以下错误！

任何帮助表示赞赏！

18:43:15,734 INFO  org.apache.flink.api.java.typeutils.TypeExtractor        - class org.joda.time.DateTime is not a valid POJO type
Exception in thread "main" java.lang.NoClassDefFoundError: kafka/producer/Partitioner
    at java.lang.ClassLoader.defineClass1(Native Method)
    at java.lang.ClassLoader.defineClass(ClassLoader.java:760)
    at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
    at java.net.URLClassLoader.defineClass(URLClassLoader.java:455)
    at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:367)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
    at java.security.AccessController.doPrivileged(Native Method)
    at java.net.URLClassLoader.findClass(URLClassLoader.java:360)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    at com.dataArtisans.flinkTraining.exercises.dataStreamJava.rideCleansing.RideCleansing.main(RideCleansing.java:51)
Caused by: java.lang.ClassNotFoundException: kafka.producer.Partitioner
    at java.net.URLClassLoader$1.run(URLClassLoader.java:372)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:361)
    at java.security.AccessController.doPrivileged(Native Method)
    at java.net.URLClassLoader.findClass(URLClassLoader.java:360)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 13 more

我为 Kafka 连接器使用了以下两个依赖项之一。

<dependency> 
<groupId>org.apache.flink</groupId>
 <artifactId>flink-connector-kafka-0.9_2.11</artifactId>        <version>1.0.0</version> 
</dependency>

或者

<dependency> 
   <groupId>org.apache.flink</groupId> 
      <artifactId>flink-connector-kafka-0.10.2</artifactId>   version>0.10.2</version>
</dependency>
4

0 回答 0