带AWS kinesis的maven Apache flink使用java获取数据。lang.NoClassDefFoundError:org/apache/flink/streaming/connectors/kinesis/flinkkinesConsumer

Question

这是我连接aws运动的代码。当我试图连接FlinkkinesConsumer时，它会抛出一个类似ClassNotFoundException的错误

import configs.AWSConfigConstants;
import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime;
import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer;
import org.apache.flink.streaming.connectors.kinesis.config.ConsumerConfigConstants;

public class StreamingJob {
    public static void main(String[] args) {
        try {
            final StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            String region = "us-east-1";
            String inputStreamName = "";
            String accesskey = "";
            String secretkey = "";
            String initPosition = "LATEST";
            String arn = "";
            
            Properties consumerConfig = new Properties();           
            consumerConfig.put(AWSConfigConstants.AWS_REGION, region);
            consumerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, accesskey);
            consumerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, secretkey);
            consumerConfig.put(AWSConfigConstants.AWS_ROLE_ARN, arn );          
            consumerConfig.put(AWSConfigConstants.STREAM_INITIAL_POSITION, initPosition);

            System.out.println("Consume config properties:");
            System.out.println(consumerConfig);
            
            DataStream<String> kinesisInputStream = sEnv.addSource(new FlinkKinesisConsumer<>
            (inputStreamName,new SimpleStringSchema(),consumerConfig));
            
            System.out.println(kinesisInputStream);
            sEnv.execute("Flink Streaming Processor");          
        } catch(Exception e) {
            System.out.println(e);
        }
    }

我得到下面这个错误

Output Screenshot

提前感谢你的帮助

Answer 1

您需要在构建中包含连接器的适当版本。类似的东西，其中2.11是scala版本，1.13.2是Flink版本

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kinesis_2.11</artifactId>
    <version>1.13.2</version>
</dependency>

有关更多信息，请参见the docs