Noclassdeffounderror试图将Kafka与Flink集成在一起时

Question

我正在尝试将 kafka和 flink互间。这个想法是消耗KAFKA队列并使用Flink转换数据。我在下面提到的示例

https://github.com/abhishek-ch/evolveml/blob/master/flink/kafka/kafka-flink-example/pom.xml

这些是我的依赖。

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-core</artifactId>
            <version>0.9.1</version>
           </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.8_2.11</artifactId>
            <version>1.3.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>1.0.0</version>
        </dependency>

我也在项目中包括Kafka和Flink类。

        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-dependency-plugin</artifactId>
            <version>3.0.2</version>
            <executions>
                <execution>
                    <id>unpack</id>
                    <!-- executed just before the package phase -->
                    <phase>prepare-package</phase>
                    <goals>
                        <goal>unpack</goal>
                    </goals>
                    <configuration>
                        <artifactItems>
                            <!-- For Flink connector classes -->
                            <artifactItem>
                                <groupId>org.apache.flink</groupId>
                                <artifactId>flink-connector-kafka-0.8_2.11</artifactId>
                                <version>1.3.2</version>
                                <type>jar</type>
                                <overWrite>false</overWrite>
                                <outputDirectory>${project.build.directory}/classes</outputDirectory>
                                <includes>org/apache/flink/**</includes>
                            </artifactItem>
                            <!-- For Kafka API classes -->
                            <artifactItem>
                                <groupId>org.apache.kafka</groupId>
                                <artifactId>kafka_2.11</artifactId>
                                <version>1.0.0</version>
                                <type>jar</type>
                                <overWrite>false</overWrite>
                                <outputDirectory>${project.build.directory}/classes</outputDirectory>
                                <includes>kafka/**</includes>
                            </artifactItem>
                        </artifactItems>
                    </configuration>
                </execution>
            </executions>
       </plugin>

我用于食用Kafka队列的Java代码是

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    Map<String, String> map = new HashMap<>();
    map.put("bootstrap.servers", kafka_server);
    map.put("zookeeper.connect", "localhost:40862");
    map.put("group.id", "test");
    map.put("topic", "data");
    // parse user parameters
    ParameterTool parameterTool = ParameterTool.fromMap(map);
    DataStream<String> messageStream = null;
    try {
        messageStream = env.addSource(new org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer082<>(
                parameterTool.getRequired("topic"),
                new SimpleStringSchema(),
                parameterTool.getProperties()));
    } catch (Exception e) {
        LOGGER.error("Error", e);
    }
    // print() will write the contents of the stream to the TaskManager's standard out stream
    // the rebelance call is causing a repartitioning of the data so that all machines
    // see the messages (for example in cases when "num kafka partitions" < "num flink operators"
    messageStream.rebalance().map(new MapFunction<String, String>() {
        private static final long serialVersionUID = -6867736771747690202L;
        @Override
        public String map(String value) throws Exception {
            LOGGER.info("============================" + value);
            return "Kafka and Flink says: " + value;
        }
    }).print();
    try {
        env.execute();
    } catch (Exception e) {
        e.printStackTrace();
    }

此代码示例来自我前面提到的GitHub项目。此代码在tomcat中部署的战争文件中运行。

运行此代码时，我会收到以下错误。

      Unrecoverable error java.util.concurrent.CompletionException: java.lang.NoClassDefFoundError: org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer082 

我提到了战争提取的课程。我试图弄清楚如何解决这个问题。任何帮助或建议都高度赞赏。

Answer 1

您应该修改flink-streaming-core依赖关系，以成为对flink-streaming-java_2.11，版本1.3.2的依赖性。（几年前，Flink-streaming-core更名为Flink-streaming-java和Flink-scaling-scala。）

此外，Flink-Connector-Kafka-0.8_2.11适用于Kafka版本0.8.x，而您将其与Kafka版本1.0.0结合使用。我建议您删除kafka_2.11的依赖性，并依靠maven延误包括kafka jar的正确版本。

Answer 2

a NoClassDefFoundError通常暗示版本/依赖关系问题，实际上您的依赖项有点混乱。

您正在从1.3.2（当前发行版）和0.9.1（相当古老的版本）中导入Flink依赖项。Flink Kafka连接器适用于Kafka 0.8，但是您可以拉动Kafka 1.0.0依赖项。