亚马逊 Kinesis Stream 创建问题

Question

我是 Amazon Kinesis 的新手。我已阅读基本文档和 Kinesis APi 以开始使用 Kinesis。所以我在 scala 中编写了一小段代码来创建流。但我面临如下异常

线程中的异常 "main" java.lang.NoSuchMethodError: org.joda.time.format.DateTimeFormatter.withZoneUTC()Lorg/joda/time/format/DateTimeFormatter; 在 com.amazonaws.auth.AWS4Signer.(AWS4Signer.java:44) 在 sun.reflect.NativeConstructorAccessorImpl.newInstance0（本机方法）在 sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57) 在 sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) 在 java.lang.reflect.Constructor.newInstance(Constructor.java:526) 在 java.lang.Class.newInstance(Class.java:379) 在 com.amazonaws.auth.SignerFactory.createSigner(SignerFactory.java:119) 在 com.amazonaws.auth.SignerFactory.lookupAndCreateSigner(SignerFactory.java:105) 在 com.amazonaws.auth.SignerFactory.getSigner(SignerFactory.java:78) 在 com.amazonaws.AmazonWebServiceClient.computeSignerByServiceRegion(AmazonWebServiceClient.java:307) 在 com.amazonaws.AmazonWebServiceClient.computeSignerByURI(AmazonWebServiceClient.java:280) 在 com.amazonaws.AmazonWebServiceClient.setEndpoint(AmazonWebServiceClient.java:160) 在 com.amazonaws.services.kinesis.AmazonKinesisClient.setEndpoint(AmazonKinesisClient.java:2102) 在 com.amazonaws.services.kinesis.AmazonKinesisClient.init(AmazonKinesisClient.java:216) 在 com.amazonaws.services.kinesis.AmazonKinesisClient.(AmazonKinesisClient.java:139) 在 com.amazonaws.services.kinesis.AmazonKinesisClient.(AmazonKinesisClient.java:116) 在 com.platalytics.platform.connectors.Kinesis.App$.create(App.scala:32) 在 com.platalytics.platform.connectors.Kinesis.App$.main(App.scala:26) 在 com.platalytics.platform.connectors.Kinesis.App.main(App.scala) 在 sun.reflect.NativeMethodAccessorImpl.invoke0（本机方法）在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:606)

这是我的一段代码

def main(args: Array[String]) {
        println("Stream creation started")
        if(create(2))
            println("Stream is created successfully")
    }
    def create(shardCount: Int): Boolean = {
        val credentials = new BasicAWSCredentials(KinesisProperties.AWS_ACCESS_KEY_ID, KinesisProperties.AWS_SECRET_KEY)

        var kinesisClient: AmazonKinesisClient = new AmazonKinesisClient(credentials)
        kinesisClient.setEndpoint(KinesisProperties.KINESIS_ENDPOINT_URL,
            KinesisProperties.KINESIS_SERVICE_NAME,
            KinesisProperties.KINESIS_REGION_ID)
        val createStreamRequest = new CreateStreamRequest()
        createStreamRequest.setStreamName(KinesisProperties.myStreamName);
        createStreamRequest.setShardCount(shardCount)
        val describeStreamRequest = new DescribeStreamRequest()
        describeStreamRequest.setStreamName(KinesisProperties.myStreamName)
        try {
            Thread.sleep(120000)
        } catch {
            case e: Exception =>
        }
        var streamStatus = "not active"
        while (!streamStatus.equalsIgnoreCase("ACTIVE")) {
            try {
                Thread.sleep(1000)
            } catch {
                case e: Exception => e.printStackTrace()
            }
            try {
                val describeStreamResponse = kinesisClient.describeStream(describeStreamRequest)
                streamStatus = describeStreamResponse.getStreamDescription.getStreamStatus
            } catch {
                case e: Exception => e.printStackTrace()
            }
        }
        if (streamStatus.equalsIgnoreCase("ACTIVE"))
            true
        else
            false
    }

这是我在 pom 文件中的依赖项

<dependency>
            <groupId>joda-time</groupId>
            <artifactId>joda-time</artifactId>
            <version>2.6</version>
        </dependency>

        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>amazon-kinesis-client</artifactId>
            <version>1.2.0</version>
        </dependency>
        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>aws-java-sdk</artifactId>
            <version>1.8.9</version>
        </dependency>

Answer 1

嗨，伙计们，我通过将 joda-time jar 文件添加到 HADOOP_CLASSPATH 找到了解决此问题的方法。现在可以使用了

亚马逊 Kinesis Stream 创建问题

amazon Kinesis Stream creation issue

scala

amazon-kinesis