Writing to s3 parquet file with Multipart Upload API

[Susmit07]

Hello Developers

We are a bit stuck, and need your assistance

We are trying to read data in hdfs using pekko hdfs connectors for parquet and upload in s3 bucket, the file format is parquet. The read is happening correctly

READ

override def readParquetFileSource(filePath: String): Source[GenericRecord, NotUsed] = {
    ugi.doAs(new PrivilegedExceptionAction[Source[GenericRecord, NotUsed]] {
      override def run(): Source[GenericRecord, NotUsed] = {
        try {
          val hadoopPath = new HadoopPath(filePath)
          hadoopConfig.setBoolean(AvroReadSupport.AVRO_COMPATIBILITY, true)
          val inputFile = HadoopInputFile.fromPath(hadoopPath, hadoopConfig)
          val reader = AvroParquetReader.builder[GenericRecord](inputFile).withConf(hadoopConfig).build()
          AvroParquetSource(reader)
        } catch {
          case ex: Exception =>
            handleParquetFileReadError(filePath, ex)
            Source.empty[GenericRecord]
        }
      }
    })
  }

WRITE

val parquetSource: Source[GenericRecord, NotUsed] = fileHandler.readParquetFileSource(filePath)
val recordToByteStringFlow: Flow[GenericRecord, ByteString, NotUsed] = Flow[GenericRecord].map(serializeRecord)

val s3Sink: Sink[ByteString, Future[MultipartUploadResult]] = S3.multipartUpload("bg0975-cef-ccmedev-data", s3key)
val uploadResult: Future[MultipartUploadResult] = parquetSource
      .via(recordToByteStringFlow)
      .runWith(s3Sink)
          
 private def serializeRecord(record: GenericRecord): ByteString = {
    val writer = new GenericDatumWriter[GenericRecord](record.getSchema)
    val outputStream = new ByteArrayOutputStream()
    val encoder: BinaryEncoder = EncoderFactory.get().binaryEncoder(outputStream, null)
    writer.write(record, encoder)
    encoder.flush()
    ByteString(outputStream.toByteArray)
  }

The file is getting uploaded but the data is getting corrupted, requesting your help. Requesting for your guidance.

[raboof]

In what way is the data corrupted? Can you verify whether the ByteString is already corrupted, or that the corruption happens later in the process?

[Susmit07]

Hey @raboof thanks for the reply, yes ByteString actually corrupts the data, I checked, I have found a way for Parquet files ... I followed this documentation

https://pekko.apache.org/docs/pekko-connectors/current/avroparquet.html

I believe for parquet files write to s3 its better to have a sample code pekko-connectors-sample repository for developers. Generic s3.multipartUpload API is not giving expected results.

// Writes a stream of GenericRecords to a Parquet file on S3 using S3A
  def streamAndWriteParquetToS3(
                                 bucket: String,
                                 key: String,
                                 parquetSource: Source[GenericRecord, NotUsed],
                                 config: Configuration
                               )(implicit system: ActorSystem[_], ec: ExecutionContext): Future[Unit] = {

    // Step 1: Define the temporary key (temporary filename)
    val tempKey = s"$key.tmp" // e.g., file.parquet.tmp
    val tempUri = new URI(s"s3a://$bucket/$tempKey")
    val tempPath = new Path(tempUri)

    // Step 2: Convert Hadoop Path to OutputFile for the temporary location
    val tempOutputFile = HadoopOutputFile.fromPath(tempPath, config)

    // Step 3: Initialize the AvroParquetWriter
    createParquetWriter(tempOutputFile, parquetSource, config).flatMap { writer =>
      parquetSource
        .via(AvroParquetFlow(writer)) // Step 4: Use the AvroParquetFlow to write the data
        .runWith(Sink.ignore)
        .flatMap { _ =>
          writer.close() // Ensure the writer is closed after successful write
          logger.info(s"Successfully written Parquet file to temporary location: s3a://$bucket/$tempKey")

          // Step 5: Move the file to its final location after successful write (Copy then Delete)
          moveTempFileToFinal(bucket, tempKey, key)
        }.recoverWith {
          case ex: Exception =>
            logger.error(s"Failed to write Parquet file to S3: ${ex.getMessage}", ex)
            writer.close()
            // Cleanup: delete the temp file if write failed
            deleteTempFile(bucket, tempKey)
            Future.failed(ex)
        }
    }.recoverWith {
      case ex: Throwable =>
        logger.error(s"Failed to create ParquetWriter: ${ex.getMessage}")
        Future.failed(ex)
    }
  }

private def createParquetWriter(
                                   outputFile: HadoopOutputFile,
                                   parquetSource: Source[GenericRecord, NotUsed],
                                   configuration: Configuration
                                 )(implicit system: ActorSystem[_], ec: ExecutionContext): Future[ParquetWriter[GenericRecord]] = {

    // Extract schema from the first record in the stream and create ParquetWriter
    val schemaFuture: Future[Schema] = parquetSource.runWith(Sink.headOption).flatMap {
      case Some(firstRecord) =>
        Future.successful(firstRecord.getSchema) // Return the schema if a record is found
      case None =>
        Future.failed(new IllegalArgumentException("No records available in the source to extract schema."))
    }

    // Once the schema is extracted, create the ParquetWriter
    schemaFuture.map { schema =>
      configuration.setBoolean(AvroReadSupport.AVRO_COMPATIBILITY, true)
      AvroParquetWriter.builder[GenericRecord](outputFile)
        .withConf(configuration)
        .withSchema(schema)
        .withWriteMode(Mode.OVERWRITE)
        .withCompressionCodec(CompressionCodecName.SNAPPY)
        .withDictionaryEncoding(true)
        .build()
    }
  }

object S3ConfigManager {

  def createS3AConfiguration: Configuration = {
    val config = new Configuration()
    config.set("fs.s3a.access.key", "")
    config.set("fs.s3a.secret.key", "")
    config.set("fs.s3a.endpoint", AppConfig.s3OverrideEndpoint)
    config.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
    config.set("fs.s3a.path.style.access", "true")
    config.set("fs.s3a.connection.ssl.enabled", "true")
    config
  }
}