S3 multipart upload for parquet

[Susmit07]

Will multipart upload works for parquet files with S3 connectors?

[raboof]

I don't see why not. Did you resolve the issue you described in #866?

[Susmit07]

Yeah issue was resolved using AvroParquetWriter but today I was able to make it work with S3MulipartUpload and as well as PutObject for small files

It matches the checksum too

[raboof]

Awesome! Can you share what you learned so we can improve the documentation?

[Susmit07]

In nutshell the doc to be updated with:

S3 PutObject API:

    // Function to stream HDFS file as a ByteString source
    private def streamHDFSFile: Source[ByteString, _] = {
      val hadoopConfig = new Configuration()
      val hdfsPath = new Path(hdfsFilePath)

      // Get the HDFS FileSystem and open the file stream
      val fs = FileSystem.get(hadoopConfig)
      HdfsSource.data(fs, hdfsPath)
    }

    // Function to get the file size from HDFS / content-length
    private def getHDFSFileSize: Long = {
      val hadoopConfig = new Configuration()
      val hdfsPath = new Path(hdfsFilePath)
      val fs = FileSystem.get(hadoopConfig)
      fs.getFileStatus(hdfsPath).getLen
    }

    // Function to upload small files using putObject with content length
    private def uploadSmallFileToS3(contentLength: Long): Future[_] = {
      val hdfsSource: Source[ByteString, _] = streamHDFSFile

      // Define the S3 putObject with content length
      val s3Headers = S3Headers.empty
      val s3PutObject = S3.putObject(
        bucket,
        bucketKey,
        hdfsSource,
        contentLength,
        ContentTypes.`application/octet-stream`,
        s3Headers
      )

      // Run the stream to upload the file using putObject
      s3PutObject.runWith(Sink.head).flatMap(objectMetadata => {
        println(s"Uploaded successfully with ETag: ${objectMetadata.eTag}")
        Future.successful(Done)
      })
    }

Configuration:

application.conf (src/resource)

s3 {
  region = "us-west-2"
  endpoint = ""
  credentials {
    access-key-id = ""
    secret-access-key = ""
  }
 # very important else bucket discovery fails, for cloudian managed s3.
  path-style-access = true
  marker {
    bucket = ${S3_MARKER_BUCKET}
  }
  sink {
    bucket = ""
  }
  ssl-enabled = true
}

Jokes apart, hope I get an acknowledgement :)

[Susmit07]

Several advantages of using S3.multipartUpload & S3.putObject

• Both the APIs support all or nothing, no dangling file visible if Process restarts
• For large files we can use Multipart Upload, and can resume upload if we can keep track of the last upload eTag
• Support for any file type.
• AvroParquetWriter can also work but its confined to Parquet files only, and need hadoop-aws library dependency to uplaod to s3. (added overhead)
• Checksum of source and target files matches, which is not possible in AvroParquetWriter, as the compression, codecs are altered

[Susmit07]

@raboof Please get it added it will help the community a lot

Take care