How to create data-frame on rocks db (SST files)

Question

We hold our documents in rocks-db. We will be syncing these rocks-db sst files to S3. I would like to create a dataframe on the SST files and later run an sql. When googled, I was not able to find any relevant connector for the same.

What is the best way to do this? Can we write a custom dataframe implementation for this? We use Spark 3.1.0 with scala 2.12

( converting rocks db to json and to parquet and then reading dataframe out of parquet is very time consuming and resource intensive --120 1-core node minutes. So I cant use this approach )

Answer 1

If you know how to parse files using plain scala code, you just need spark to distribute it on executors:

1. List file names in your s3 bucket, It will result in Seq[String]

2. Transform them to dataset/dataframe (spark.createDataset)

3. Do mapPartitions operation

4. Inside mapPartitions, init s3Client, read file content of files using it. Parse using plain scala code, output as case class

5. Output of mapPratitions will be dataset of parsed files

upd:

that is sample of code

I do not know what is rock db files and how to read them via plain scala code

Assuming you can read them with plain scala, without spark

next code sample can be used to parallelize read/parse with spark

  val session: SparkSession = SparkSession.builder().config(new SparkConf().setMaster("local")).getOrCreate()
  
  import session.implicits._

  def listS3Files(bucket: String, prefix: String): Seq[String] = {
    val s3Client = AmazonS3ClientBuilder.standard.build // add credentials, region if needed
    val listObjectsRequest = (new ListObjectsV2Request)
      .withBucketName(bucket)
      .withPrefix(prefix)
    s3Client.listObjectsV2(listObjectsRequest).getObjectSummaries.asScala.map(_.getKey).filter(p => !p.endsWith("/"))
  }
  
  val bucket = "your-bucket"
  val files = listS3Files(bucket, "test")

  val result = files
    .toDS()
    .mapPartitions(fs => {
      val s3Client = AmazonS3ClientBuilder.standard.build // add credentials, region if needed
      // here got content as string, in your case you need to read rock db files and parse them to case class
      val content = fs.map(f => IOUtils.toString(s3Client.getObject(bucket, f).getObjectContent, StandardCharsets.UTF_8))
      content
    }).cache()