Невозможно получить accessKey и SecretKey от конф. Hadoop

Question

У меня есть код, как показано ниже -

import java.io.InputStream
import java.net.URI
import java.util

import com.amazonaws.auth.AWSCredentialsProviderChain
import com.amazonaws.{ClientConfiguration, Protocol}
import com.amazonaws.services.s3.AmazonS3Client
import com.amazonaws.services.s3.model._
import org.apache.hadoop.conf.{Configuration => HadoopConfiguration}
import org.apache.hadoop.fs.{Path => HadoopPath}
import org.apache.hadoop.fs.s3a.{BasicAWSCredentialsProvider, S3AFileSystem}
import com.amazonaws.services.s3.model.ObjectListing

import scala.annotation.tailrec

object FileOperation {

  val uri = new URI("s3a://bucket-name/prefixKey/file.json")
  val fs: S3AFileSystem = new S3AFileSystem()

  def getAWSClient: AmazonS3Client = {
    val conf: HadoopConfiguration = new HadoopConfiguration(true)
    val awsConf: ClientConfiguration = new ClientConfiguration()

    val secureConnections: Boolean = conf.getBoolean("fs.s3a.connection.ssl.enabled", false)
    awsConf.setProtocol(if (secureConnections) Protocol.HTTPS else Protocol.HTTP)

    val accessKey: String = conf.get("fs.s3a.access.key", null.asInstanceOf[String])
    val secretKey: String = conf.get("fs.s3a.secret.key", null.asInstanceOf[String])

    println(s"inside getAWSClient accessKey -> $accessKey ; secretKey -> $secretKey")

    val credentials = new AWSCredentialsProviderChain(new BasicAWSCredentialsProvider(accessKey, secretKey))

    val s3: AmazonS3Client = new AmazonS3Client(credentials, awsConf)
    s3.setEndpoint(conf.get("fs.s3a.endpoint", null.asInstanceOf[String]))
    s3
  }

  def getEntries(recursive: Boolean): Seq[URI] = {

    @tailrec
    def collectEntries(summaries: util.Iterator[S3ObjectSummary], collected: Seq[HadoopPath]): Seq[HadoopPath] = {
      if (summaries.hasNext) {
        val summary: S3ObjectSummary = summaries.next()
        val newPath: String = "s3a://" + summary.getBucketName + "/" + summary.getKey

        collectEntries(summaries, collected :+ {
          fs.initialize(new URI(newPath), new HadoopConfiguration(true))
          new HadoopPath(new URI(newPath))
        })
      } else collected
    }

    val prefixKey: String = if (!(uri.getScheme != null && uri.getPath.isEmpty)) uri.getPath.substring(1) else ""

    val objects: ObjectListing = getAWSClient.listObjects(uri.getHost, prefixKey)

    if (objects.getObjectSummaries.isEmpty) throw new java.nio.file.NoSuchFileException(uri.toString)
    else {
      collectEntries(objects.getObjectSummaries.iterator(), Seq.empty).map(path => path.toUri)
    }

  }

  def asStream: InputStream = {
    val prefixKey1 = if (!(uri.getScheme != null && uri.getPath.isEmpty)) uri.getPath.substring(1) else ""
    val s3object: S3Object = getAWSClient.getObject(new GetObjectRequest(uri.getHost, prefixKey1))
    s3object.getObjectContent
  }

}

вызов функции getAWSClient в функции getEntries работает, но вызов функции asStream получает пустые ключи доступа и секретный ключ. Функция getEntries используется для вывода списка файлов в папке, тогда как функция asStream возвращает входной поток таких файлов, который используется для создания BufferedSource, а затем содержимое читается.

Ниже приведен файл hadoop core-ste.xml. Может ли кто-нибудь помочь мне понять, почему в функции asStream секретные ключи и права доступа не отображаются.

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Sample config for Hadoop S3A client. -->
<configuration>

    <property>
        <name>fs.s3a.access.key</name>
        <value>xxxxxxxx</value>
    </property>
    <property>
        <name>fs.s3a.secret.key</name>
        <value>yyyyyyy</value>
    </property>
    <property>
        <name>fs.s3a.connection.ssl.enabled</name>
        <value>false</value>
    </property>
    <property>
        <name>fs.s3a.endpoint</name>
        <value>dev:80</value>
    </property>
    <property>
        <name>fs.s3a.imp</name>
        <value>org.apache.hadoop.fs.s3a.S3A</value>
    </property>
    <property>
        <name>fs.s3a.path.style.access</name>
        <value>true</value>
    </property>
</configuration>

Answer 1

1. разъем S3A не раскрывает своих секретов, потому что они, в общем-то, "секретные".
2. И класс S3AFileSystem не ожидает initialize() быть вызванным более одного раза. Дорогие вещи, такие как пулы потоков и менеджер переноса AWS, создаются и очищаются только в FileSystem.close() вызов. Ваш код пропустит это. Если бы мы знали, что люди пытаются это сделать (и теперь мы делаем!), Мы добавим проверку и быстро провалимся.

Если вы позвоните FileSystem.listFiles(path, true) вы получаете рекурсивный список всех объектов по пути с одним HTTP-запросом из S3 на тысячу записей-потомков; это то, что вы делаете с listObjects.

О, и если вы хотите, чтобы конфигурация использовалась ФС, позвоните fs.getConf(), Это будет содержать секреты, если они были переданы в файле XML. Если они сохранены в файлах JCECKs или другом безопасном хранилище, его немного сложно подобрать.