Filtering dataframe based on dynamic value in Spark/Scala

Question

I have a json in the following format :

{"Request": {"TrancheList": {"Tranche": [{"TrancheId": "500192163","OwnedAmt": "26500000",    "Curr": "USD" }, {  "TrancheId": "500213369", "OwnedAmt": "41000000","Curr": "USD"}]},"FxRatesList": {"FxRatesContract": [{"Currency": "CHF","FxRate": "0.97919983706115"},{"Currency": "AUD", "FxRate": "1.2966804979253"},{ "Currency": "USD","FxRate": "1"},{"Currency": "SEK","FxRate": "8.1561012531034"},{"Currency": "NOK", "FxRate": "8.2454981641398"},{"Currency": "JPY","FxRate": "111.79999785344"},{"Currency": "HKD","FxRate": "7.7568025218916"},{"Currency": "GBP","FxRate": "0.69425159677867"}, {"Currency": "EUR","FxRate": "0.88991723769689"},{"Currency": "DKK", "FxRate": "6.629598372301"}]},"isExcludeDeals": "true","baseCurrency": "USD"}}

I am trying to get the Fxrate value for a Currency which is equal to baseCurrency Tag

I am reading the json from hdfs cluster

 val hdfsRequest = spark.read.json("localhost/user/request.json")
val baseCurrency = hdfsRequest.select("Request.baseCurrency")
var fxRates = hdfsRequest.select("Request.FxRatesList.FxRatesContract")
val fxRatesDF = fxRates.select(explode(fxRates("FxRatesContract"))).toDF("FxRatesContract").select("FxRatesContract.Currency", "FxRatesContract.FxRate").filter($"Currency=baseCurrency")

The error that i got running this line of code is :

org.apache.spark.sql.AnalysisException: cannot resolve '`Currency=baseCurrency`' given input columns: [Currency, FxRate];

How do i specify varaible baseCurrency in the filter expression of dataframe in Scala/Spark?

Thanks

Shivansh · Accepted Answer · 2016-11-17 08:56:06Z

4

If base currency is just a single value then What you can do is :

val hdfsRequest = spark.read.json("localhost/user/request.json")
val baseCurrency = hdfsRequest.select("Request.baseCurrency")
  .map(_.getString(0)).collect.headOption
var fxRates = hdfsRequest.select("Request.FxRatesList.FxRatesContract")
val fxRatesDF = fxRates.select(explode(fxRates("FxRatesContract")))
  .toDF("FxRatesContract")
  .select("FxRatesContract.Currency", "FxRatesContract.FxRate")
  .filter($"Currency"===baseCurrency.fold(-1D)(identity))

edited Nov 17, 2016 at 8:56

answered Nov 17, 2016 at 8:04

Shivansh

3,57426 silver badges47 bronze badges

Sign up to request clarification or add additional context in comments.

5 Comments

user3290807 Over a year ago

i am getting the following error :java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Double.Is there any cleaner way to implement this .baseCurrency varaible is a string value like EUR or USD ,why should we converting it to double?

Shivansh Over a year ago

I didn't know of which type it is use String in place of Double then , I have updated my answer

user3290807 Over a year ago

fold(-1D)(identity) still gives a issue

user3290807 Over a year ago

i used the following code now:val fxRatesDF = fxRates.select(explode(fxRates("FxRatesContract"))).toDF("FxRatesContract").select("FxRatesContract.Currency", "FxRatesContract.FxRate").filter($"Currency"===baseCurrency.get).first() but it gives me a output as [USD,1] .But i am interested in grabbing the value 1 from this row

Shivansh Over a year ago

@user3290807: Try to do do something foryourself first, that would be beneficial for you. If you find any problem , post it as a different ques and accept this answer.

Collectives™ on Stack Overflow

Filtering dataframe based on dynamic value in Spark/Scala

1 Answer 1

5 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

5 Comments

Your Answer

Sign up or log in

Post as a guest

Related