Scala Programming Projects

Implementing httpToDomainTransactions

This implementation uses select to avoid an extra serialization/deserialization. Add the following function in BatchProducer:

  def httpToDomainTransactions(ds: Dataset[HttpTransaction]):     Dataset[Transaction] = {    import ds.sparkSession.implicits._    ds.select(      $"date".cast(LongType).cast(TimestampType).as("timestamp"),      $"date".cast(LongType).cast(TimestampType).       cast(DateType).as("date"),      $"tid".cast(IntegerType),      $"price".cast(DoubleType),      $"type".cast(BooleanType).as("sell"),      $"amount".cast(DoubleType))      .as[Transaction]  }

We use cast to convert the string columns into the appropriate types. For converting in to TimeStampType, we have to first convert in to LongType, and for converting in to DateType