CODEWITHSUNDEEP
javamavenapache-flinkamazon-kinesis-analytics
Sanjaybabu
Sanjaybabu

Reputation: 1

Apache flink with AWS kinesis consume data-Getting java.lang.NoClassDefFoundError: org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumer

Here is my code for connecting aws kinesis. When I'm trying to connect with FlinkKinesisConsumer it's throwing an error like Classnotfound exception.

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

I'm getting this below error

Output Screenshot

Thanks in advance for your help.

Upvotes: 0

Views: 1370

Answers (1)

David Anderson
David Anderson

Reputation: 43439

You need to include the appropriate version of the connector in your build. Something like this, where 2.11 is the scala version, and 1.13.2 is the Flink version.

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

For more info, see the docs.

Upvotes: 1

Related Questions