spark StreamingListener cloudWatch集成

Question

嘿，我正在尝试通过StreamingListener结束对cloudWatch的一些结构化指标

如下所示：

class MyStreamingListener()
extends StreamingListener{

 override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted):Unit={
 val cloudWatch = new AmazonCloudWatchClient(new BasicAWSCredentials(awsAccessKeyId, awsSecretKey))
  cloudWatch.setEndpoint("monitoring.eu-west-1.amazonaws.com")
  val putMetricDataRequest = new PutMetricDataRequest()
    putMetricDataRequest.setNamespace("my-name-space")
    val metricDatum = new MetricDatum().withMetricName("test")
    metricDatum.setValue(batchCompleted.batchInfo.numRecords)
    metricDatum.setUnit(StandardUnit.fromValue("Milliseconds"))
    putMetricDataRequest.getMetricData.add(metricDatum)
    cloudWatch.putMetricData(putMetricDataRequest)
}
}

val streamingContext: StreamingContext = new StreamingContext(spark.sparkContext, Seconds(2))
  streamingContext.addStreamingListener(new LoadIndexStreamingListener)

  val dstream = KinesisUtils.createStream(
      streamingContext, "this-is-just-a-test", "my-stream", "kinesis.eu-west-1.amazonaws.com",
      "eu-west-1", InitialPositionInStream.LATEST, Seconds(2), StorageLevel.MEMORY_AND_DISK_2)
      .map(byteArray => new String(byteArray))
  dstream.print()
  streamingContext.start()
  streamingContext.awaitTermination()

当我在我的集群上使用spark-shell进行一些测试时，它工作正常，指标被发送到CloudWacth

但是，当我使用sbt clean assembly将代码打包到jar中并使用spark submit运行它时，我得到了以下错误：

java.lang.NoSuchMethodError: com.amazonaws.services.cloudwatch.AmazonCloudWatchClient.putMetricData(Lcom/amazonaws/services/cloudwatch/model/PutMetricDataRequest;)Lcom/amazonaws/services/cloudwatch/model/PutMetricDataResult;

spark-submit --class com.me.sparkTest.App --master local[4]  --packages org.apache.spark:spark-streaming-kinesis-asl_2.11:2.1.0,com.amazonaws:amazon-kinesis-client:1.7.2 clowdwatch-spark-test-assembly-1.0.jar

你知道是什么原因导致它在使用spark-submit时失败吗？

Answer 1

该错误是由于编译后的类具有与AmazonCloudWatchClient.putMetricData集群中可用的运行时库不同的签名而导致的。

这个问题的解决方案是创建一个具有以下依赖项的uber jar

<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-java-sdk</artifactId>
    <version>1.10.75.1</version>
</dependency>