Spark transformation for timeseries/tick data set

Question

We have table in hive which stores trading orders data for each end of day as order_date. Other important columns are product, contract, price(price of order placed), ttime (transaction time) status (insert,update or remove) price (price of order)

We have to build a charting table in in tick data fashion from the main table with the max and min price orders for each row(order) from the morning when market opened till that time. i.e for a given order we would have 4 columns populated as maxPrice(max price till now), maxpriceOrderId(orderid of the max price), minPrice and minPriceOrderId
This has to be for each product, contract i.e max and min prices for among that product,contract.

While calculating these values we need to exclude all closed orders from the aggregation. i.e max and min over all order prices till now excluding orders with status as "Remove"

We are using : Spark 2.2 and input data format is parquet. Input records

Output Records

To give a simple SQL view - the problem work out with a self join and would look like this : With a ordered data set on ttime we have to get the max and min prices for a particular product,contract for each row(order) from the morning till that order's time. This will run on for each eod (order_date) data set in batch:

select mainSet.order_id,    mainSet.product,mainSet.contract,mainSet.order_date,mainSet.price,mainSet.ttime,mainSet.status,
max(aggSet.price) over (partition by mainSet.product,mainSet.contract,mainSet.ttime) as max_price,
first_value(aggSet.order_id) over (partition by mainSet.product,mainSet.contract,mainSet.ttime) order by (aggSet.price desc,aggSet.ttime desc ) as maxOrderId
min(aggSet.price) over (partition by mainSet.product,mainSet.contract,mainSet.ttime) as min_price as min_price
first_value(aggSet.order_id) over (partition by mainSet.product,mainSet.contract,mainSet.ttime) order by (aggSet.price ,aggSet.ttime) as minOrderId
from order_table mainSet 
join order_table aggSet
ON (mainSet.produuct=aggSet.product,
mainSet.contract=aggSet.contract,
mainSet.ttime>=aggSet.ttime,
aggSet.status <> 'Remove')

Writing in Spark :

We started with spark sql like below:

val mainDF: DataFrame= sparkSession.sql("select * from order_table where order_date ='eod_date' ")

  val ndf=mainDf.alias("mainSet").join(mainDf.alias("aggSet"),
        (col("mainSet.product")===col("aggSet.product")
          && col("mainSet.contract")===col("aggSet.contract")
          && col("mainSet.ttime")>= col("aggSet.ttime")
          && col("aggSet.status") <> "Remove")
        ,"inner")
        .select(mainSet.order_id,mainSet.ttime,mainSet.product,mainSet.contract,mainSet.order_date,mainSet.price,mainSet.status,aggSet.order_id as agg_orderid,aggSet.ttime as agg_ttime,price as agg_price) //Renaming of columns

  val max_window = Window.partitionBy(col("product"),col("contract"),col("ttime"))
  val min_window = Window.partitionBy(col("product"),col("contract"),col("ttime"))
  val maxPriceCol = max(col("agg_price")).over(max_window)
  val minPriceCol = min(col("agg_price")).over(min_window)
  val firstMaxorder = first_value(col("agg_orderid")).over(max_window.orderBy(col("agg_price").desc, col("agg_ttime").desc))
  val firstMinorder = first_value(col("agg_orderid")).over(min_window.orderBy(col("agg_price"), col("agg_ttime")))      


  val priceDF=  ndf.withColumn("max_price",maxPriceCol)
                    .withColumn("maxOrderId",firstMaxorder)
                    .withColumn("min_price",minPriceCol)
                    .withColumn("minOrderId",firstMinorder)

    priceDF.show(20)

Volume Stats:

avg count 7 Million records avg count for each group (product,contract)= 600K

The job is running for hours and just not finishing.I have tried increasing memory and other parameters but no luck. Job is getting stuck and many times I am getting memory issues Container killed by YARN for exceeding memory limits. 4.9 GB of 4.5 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead

ANOTHER APPROACH :

Doing repartitioning on our lowest group columns(product and contract) and then sorting within partitions on time so that we receive each row ordered on time for the mapPartition function.

Executing mappartition while maintaining a collection (key as order_id and price as value) at partition level to calculate the max and min price and their orderids.

We will keep removing the orders with status as "Remove" from collection as and when we recieve them. Once collection is updated for a given row within mapparition we can calculate max and min values from collection and return updated row.

val mainDF: DataFrame= sparkSession.sql("select order_id,product,contract,order_date,price,status,null as maxPrice,null as maxPriceOrderId,null as minPrice,null as minPriceOrderId from order_table where order_date ='eod_date' ").repartitionByRange(col("product"),col("contract"))

case class summary(order_id:String ,ttime:string,product:String,contract :String,order_date:String,price:BigDecimal,status :String,var maxPrice:BigDecimal,var maxPriceOrderId:String ,var minPrice:BigDecimal,var minPriceOrderId String)

val summaryEncoder = Encoders.product[summary]
val priceDF= mainDF.as[summary](summaryEncoder).sortWithinPartitions(col("ttime")).mapPartitions( iter => {
    //collection at partition level
    //key as order_id and value as price
    var priceCollection = Map[String, BigDecimal]()

    iter.map( row => {
        val orderId= row.order_id
        val rowprice= row.price

        priceCollection = row.status match {
                            case "Remove" => if (priceCollection.contains(orderId)) priceCollection -= orderId
                            case _ => priceCollection += (orderId -> rowPrice)
                         }

        row.maxPrice = if(priceCollection.size > 0) priceCollection.maxBy(_._2)._2  // Gives key,value tuple from collectin for  max value )
        row.maxPriceOrderId = if(priceCollection.size > 0) priceCollection.maxBy(_._2)._1

        row.minPrice =  if(priceCollection.size > 0) priceCollection.minBy(_._2)._2   // Gives key,value tuple from collectin for  min value )
        row.minPriceOrderId = if(priceCollection.size > 0) priceCollection.minBy(_._2)._1

      row

    })
  }).show(20)

This is running fine and finishing it within 20 mins for smaller data sets but I found for a 23 mill records (having 17 diff product and contracts) the result seems not correct. I can see the data from one partition(input split) of mappartition is going another partition and thus messing up the values.

--> Can we achieve a situation in which I can guaranty that each mappartition task would get all the data for functional key (product and contract) here.. As I know ,mappartition executes function on each spark partition(similar to input splits in map reduce) so how can I force spark to create inputsplits/partitions having all of the values for that product and contract group.

--> Is there any other approach to this problem ?

Would really appreciate the help as we are stuck here.

Answer 1

Edit: Here's an article on why many small files are bad

Why is poorly compacted data bad? Poorly compacted data is bad for Spark applications in the sense that it is extremely slow to process. Continuing with our previous example, anytime we want to process a day’s worth of events we have to open up 86,400 files to get to the data. This slows down processing massively because our Spark application is effectively spending most of its time just opening and closing files. What we normally want is for our Spark application to spend most of its time actually processing the data. We’ll do some experiments next to show the difference in performance when using properly compacted data as compared to poorly compacted data.

---

I bet if you properly partitioned your source data to however you're joining AND get rid of all those windows, you'd end up in a much better place.

Every time you hit partitionBy, you are forcing a shuffle and every time you hit orderBy, you are forcing an expensive sort.

I would suggest you take a look at the Dataset API and learn some groupBy and flatMapGroups/reduce/sliding for O(n) time computation. You can get your min/max in one pass.

Furthermore, it sounds like your driver is running out of memory due to the many little files problem. Try to compact your source data as much as possible and properly partition your tables. In this particular case I would suggest a partitioning by order_date (maybe daily?) then sub partitions of product and contract.

Here's a snippet that took me about 30 minutes to write and probably runs exponentially better than your windowing function. It should run in O(n) time but it doesn't make up for if you have a many small files problem. Let me know if anything's missing.

import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession}
import scala.collection.mutable

case class Summary(
  order_id: String,
  ttime: String,
  product: String,
  contract: String,
  order_date: String,
  price: BigDecimal,
  status: String,
  maxPrice: BigDecimal = 0,
  maxPriceOrderId: String = null,
  minPrice: BigDecimal = 0,
  minPriceOrderId: String = null
)

class Workflow()(implicit spark: SparkSession) {

  import MinMaxer.summaryEncoder

  val mainDs: Dataset[Summary] =
    spark.sql(
      """
        select order_id, ttime, product, contract, order_date, price, status
        from order_table where order_date ='eod_date'
      """
    ).as[Summary]

  MinMaxer.minMaxDataset(mainDs)
}

object MinMaxer {

  implicit val summaryEncoder: Encoder[Summary] = Encoders.product[Summary]
  implicit val groupEncoder: Encoder[(String, String)] = Encoders.product[(String, String)]

  object SummaryOrderer extends Ordering[Summary] {
    def compare(x: Summary, y: Summary): Int = x.ttime.compareTo(y.ttime)
  }

  def minMaxDataset(ds: Dataset[Summary]): Dataset[Summary] = {
    ds
      .groupByKey(x => (x.product, x.contract))
      .flatMapGroups({ case (_, t) =>
        val sortedRecords: Seq[Summary] = t.toSeq.sorted(SummaryOrderer)

        generateMinMax(sortedRecords)
      })
  }

  def generateMinMax(summaries: Seq[Summary]): Seq[Summary] = {
    summaries.foldLeft(mutable.ListBuffer[Summary]())({case (b, summary) =>

      if (b.lastOption.nonEmpty) {
        val lastSummary: Summary = b.last

        var minPrice: BigDecimal = 0
        var minPriceOrderId: String = null
        var maxPrice: BigDecimal = 0
        var maxPriceOrderId: String = null

        if (summary.status != "remove") {
          if (lastSummary.minPrice >= summary.price) {
            minPrice = summary.price
            minPriceOrderId = summary.order_id
          } else {
            minPrice = lastSummary.minPrice
            minPriceOrderId = lastSummary.minPriceOrderId
          }

          if (lastSummary.maxPrice <= summary.price) {
            maxPrice = summary.price
            maxPriceOrderId = summary.order_id
          } else {
            maxPrice = lastSummary.maxPrice
            maxPriceOrderId = lastSummary.maxPriceOrderId
          }

          b.append(
            summary.copy(
              maxPrice = maxPrice,
              maxPriceOrderId = maxPriceOrderId,
              minPrice = minPrice,
              minPriceOrderId = minPriceOrderId
            )
          )
        } else {
          b.append(
            summary.copy(
              maxPrice = lastSummary.maxPrice,
              maxPriceOrderId = lastSummary.maxPriceOrderId,
              minPrice = lastSummary.minPrice,
              minPriceOrderId = lastSummary.minPriceOrderId
            )
          )
        }
      } else {
        b.append(
          summary.copy(
            maxPrice = summary.price,
            maxPriceOrderId = summary.order_id,
            minPrice = summary.price,
            minPriceOrderId = summary.order_id
          )
        )
      }

      b
    })
  }
}

Answer 2

The method that you have used to repartition the data repartitionByRange partitions the data on these column expressions but does a range partitioning. What you want is hash partitioning on these columns.

Change the method to repartition and pass these columns to it and it should ensure that the same value groups end up in one partition.