Как преобразовать CSV-файл в S3 ведро в RDD

Question

Я довольно новичок в этой теме, поэтому любая помощь будет высоко оценена.

Я пытаюсь прочитать CSV-файл, который хранится в S3-контейнере, и преобразовать его данные в RDD для непосредственной работы с ним.это без необходимости создавать файл локально.

До сих пор я был в состоянии загрузить файл с помощью AmazonS3ClientBuilder, но единственное, что у меня есть, это иметь содержимое файла в S3ObjectInputStream, и я 'Я не могу работать с его содержимым.

val bucketName = "bucket-name"

val credentials = new BasicAWSCredentials(
   "acessKey",
   "secretKey"
);

val s3client = AmazonS3ClientBuilder
    .standard()
    .withCredentials(new AWSStaticCredentialsProvider(credentials))
    .withRegion(Regions.US_EAST_2)
    .build();

val s3object = s3client.getObject(bucketName, "file-name.csv")
val inputStream = s3object.getObjectContent()
....

Я также пытался использовать BufferedSource для работы с ним, но однажды я не знаю, как преобразовать его в фрейм данных или RDD для работы.с ним.

val myData = Source.fromInputStream(inputStream)
....

Answer 1

В конце я смог получить результаты, которые искал, чтобы взглянуть на https://gist.github.com/snowindy/d438cb5256f9331f5eec

Answer 2

Вы можете сделать это с файловой системой S3A, предоставленной в модуле Hadoop-AWS:

1. Добавить эту зависимость https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-aws
2. Либо определите <property><name>fs.s3.impl</name><value>org.apache.hadoop.fs.s3a.S3AFileSystem</value></property> в core-site.xml, либо добавьте .config("fs.s3.impl", classOf[S3AFileSystem].getName) в SparkSession builder
3. Для доступа к S3 используйте spark.read.csv("s3://bucket/key"). Если вы хотите RDD, который был задан spark.read.csv("s3://bucket/key").rdd