title: Unable to read data from S3 in Pyspark on EC2 (standalone setup)
body: <p>I am trying to read the CSV file (or any file for that matter) from S3 (SSE_KMS encrypted) into pyspark setup locally on EC2 machine. I am getting different type of error with different combination. I am able to download the file using AWS cli shown below:</p>
  <p><code>aws s3 ls s3://xxxx/ --profile xx </code></p>
  <p>The above command seems to be working fine. Please see the pyspark code below for which I am getting error (this I am testing in pyspark shell):</p>
  <pre><code>spark.conf.set(&quot;fs.s3a.awsAccessKeyId&quot;,&quot;XXXXXXXXXXXXXX&quot;)
  spark.conf.set(&quot;fs.s3a.awsSecretAccessKey&quot;, &quot;XXXXXXXXXXXXXXXXXXXXXXXXXX&quot;)
  df = spark.read.format(&quot;csv&quot;).load(&quot;s3a://bucket/test.csv&quot;)
  </code></pre>
  <p>The error I am getting below is:</p>
  <pre><code>21/05/10 04:35:45 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0)
  com.amazonaws.services.s3.model.AmazonS3Exception: Status Code: 400, AWS Service: Amazon S3, AWS Request ID: xxxx, AWS Error Code: InvalidArgument, AWS Error Message: Requests specifying Server Side Encryption with AWS KMS managed keys require AWS Signature Version 4., S3 Extended Request ID: xxx
          at com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:798)
          at com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:421)
          at com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:232)
          at com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:3528)
          at com.amazonaws.services.s3.AmazonS3Client.getObject(AmazonS3Client.java:1111)
          at org.apache.hadoop.fs.s3a.S3AInputStream.reopen(S3AInputStream.java:91)
          at org.apache.hadoop.fs.s3a.S3AInputStream.openIfNeeded(S3AInputStream.java:62)
          at org.apache.hadoop.fs.s3a.S3AInputStream.read(S3AInputStream.java:156)
          at java.io.DataInputStream.read(DataInputStream.java:149)
          at org.apache.hadoop.mapreduce.lib.input.UncompressedSplitLineReader.fillBuffer(UncompressedSplitLineReader.java:62)
          at org.apache.hadoop.util.LineReader.readDefaultLine(LineReader.java:216)
          at org.apache.hadoop.util.LineReader.readLine(LineReader.java:174)
          at org.apache.hadoop.mapreduce.lib.input.UncompressedSplitLineReader.readLine(UncompressedSplitLineReader.java:94)
          at org.apache.hadoop.mapreduce.lib.input.LineRecordReader.skipUtfByteOrderMark(LineRecordReader.java:144)
          at org.apache.hadoop.mapreduce.lib.input.LineRecordReader.nextKeyValue(LineRecordReader.java:184)
          at org.apache.spark.sql.execution.datasources.RecordReaderIterator.hasNext(RecordReaderIterator.scala:39)
          at org.apache.spark.sql.execution.datasources.HadoopFileLinesReader.hasNext(HadoopFileLinesReader.scala:69)
          at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
          at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409)
          at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:101)
          at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.nextIterator(FileScanRDD.scala:181)
          at org.apache.spark.sql.execution.datasources.FileScanRDD$$anon$1.hasNext(FileScanRDD.scala:101)
          at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
          at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
          at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636)
          at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
          at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
          at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
          at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
          at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
          at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
          at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
          at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
          at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
          at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
          at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
          at org.apache.spark.scheduler.Task.run(Task.scala:123)
          at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
          at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
          at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
          at java.lang.Thread.run(Thread.java:748)
  </code></pre>
  <p>So I assumed this was related to KMS but when I am using the below code:</p>
  <pre><code>spark.conf.set(&quot;fs.s3.awsAccessKeyId&quot;,&quot;xxxxxxxxxxxxxxxxxxxxxxxx&quot;)
  spark.conf.set(&quot;fs.s3.awsSecretAccessKey&quot;, &quot;xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx&quot;)
  spark.conf.set(&quot;fs.s3.session.token&quot;,&quot;xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx&quot;)
  
  spark.sparkContext.setSystemProperty(&quot;com.amazonaws.services.s3.enableV4&quot;, &quot;true&quot;)
  spark.conf.set(&quot;fs.s3.server-side-encryption-algorithm&quot;,&quot;SSE-KMS&quot;)
  spark.conf.set(&quot;fs.s3.server-side-encryption.key&quot;,&quot;value&quot;)
  df = spark.read.format(&quot;csv&quot;).load(&quot;s3://bucket/test.csv&quot;)
  </code></pre>
  <p>I am getting the below error:</p>
  <pre><code>py4j.protocol.Py4JJavaError: An error occurred while calling o32.load.
  : org.apache.hadoop.fs.s3.S3Exception: org.jets3t.service.S3ServiceException: Service Error Message. -- ResponseCode: 403, ResponseStatus: Forbidden, XML Error Message: &lt;?xml version=&quot;1.0&quot; encoding=&quot;UTF-8&quot;?&gt;&lt;Error&gt;&lt;Code&gt;InvalidAccessKeyId&lt;/Code&gt;&lt;Message&gt;The AWS Access Key Id you provided does not exist in our records.&lt;/Message&gt;&lt;AWSAccessKeyId&gt;XXXXXXXXXXX&lt;/AWSAccessKeyId&gt;&lt;RequestId&gt;xxx&lt;/RequestId&gt;&lt;HostId&gt;xxx//xxxxxxxxxxxxxxxxxxxxxxxx=&lt;/HostId&gt;&lt;/Error&gt;
  </code></pre>
  <p>SPARK VERSION: spark-2.4.7-bin-hadoop2.7
  The version of external jars (added to $SPARK_HOME/jars) I am using :</p>
  <pre><code>hadoop-aws-2.7.7.jar
  jets3t-0.9.4.jar
  spark-xml_2.11-0.4.1.jar
  aws-java-sdk-1.7.4.jar
  aws-java-sdk-s3-1.11.1014.jar
  </code></pre>
  <p>I have been working on this but could not find the workaround. I know there are many questions in Stack Overflow itself on this subject but none of them have worked for me.</p>
  <p>Any help would be really appreciated.</p>
  <p>UPDATE:</p>
  <p>Below is the policy setup for my bucket</p>
  <pre><code>{
      &quot;Version&quot;: &quot;******&quot;,
      &quot;Statement&quot;: [
  {
              &quot;Sid&quot;: &quot;xxxxxxxxxxxx&quot;,
              &quot;Action&quot;: [
                  &quot;s3:GetBucketLocation&quot;,
                  &quot;s3:PutObject&quot;,
                  &quot;s3:GetObject&quot;,
                  &quot;s3:GetEncryptionConfiguration&quot;,
                  &quot;s3:AbortMultipartUpload&quot;,
                  &quot;s3:ListMultipartUploadParts&quot;,
                  &quot;s3:ListBucket&quot;,
                  &quot;s3:ListBuckets&quot;,
                  &quot;s3:ListBucketMultipartUploads&quot;,
                  &quot;s3:DeleteObject&quot;,
                  &quot;s3:DeleteObjects&quot;
              ],
              &quot;Effect&quot;: &quot;Allow&quot;,
              &quot;Resource&quot;: [
                  &quot;arn:aws:s3:::xxxxxxxxxx-*&quot;,
                  &quot;arn:aws:s3:::xxxxxxxxxx-*/*&quot;
              ]
          },
          ]
      }
  </code></pre>
  <p>I also tried with the below settings (IAM role based credentials) but got the error.</p>
  <p>The code:</p>
  <pre><code>spark.conf.set(&quot;fs.s3.assumed.role.arn&quot;, &quot;xxxxxxxxxxxxxxxxxxxxxxxxxxxx&quot;)
  spark.read.csv(&quot;s3://&quot;)
  </code></pre>
  <p>The error:</p>
  <pre><code>py4j.protocol.Py4JJavaError: An error occurred while calling o37.load.
  : java.lang.IllegalArgumentException: AWS Access Key ID and Secret Access Key must be specified as the username or password (respectively) of a s3 URL, orby setting the fs.s3.awsAccessKeyId or fs.s3.awsSecretAccessKey properties (respectively).
  </code></pre>
  <p>And if I use <code>fs.s3</code> instead of <code>fs.s3a</code>. I get below error:</p>
  <pre><code>    py4j.protocol.Py4JJavaError: An error occurred while calling o37.load.
  : java.lang.NumberFormatException: For input string: &quot;100M&quot;
  </code></pre>