Category Archives: Spark

This is the final post in the Predicting Global Energy Demand using Spark series. The series has the following posts.

Problem:

Assuming there was full day of outage, Calculate the Revenue Loss for a particular day next year by finding the Average Revenue per day (ARPD) of the household(Use the below tariff plan).

Time of Use (TOU) tariff plan:

Brief Solution:

We need to find the average cost per day in a year which will be the revenue loss.
To solve this problem we need to aggregate the data on hourly basis. Based on the tariff plan calculate the cost per hour. Aggregate this cost into daily basis. Add the cost for every day in a year and get the average by dividing it with the total number of days in that year. This average represents the revenue loss in case of outage.

Getting per hour revenue:

To get the revenue per hour you can refer part 3 hourly aggregation. After getting the hourly data submeter readings will be multiplied by tariff applied for that particular hour. This will give us the cost on every hour.

def getPerHourRevenue(inputRDD:RDD[Record]):RDD[Record] = {
    val dataAggregator = new DataAggregator()
    val hourlyRDD = dataAggregator.hourlyAggregator(inputRDD)
    hourlyRDD.map(value => {
      var ((date, hour), record) = value
      val hourOfDay = record.hourofDay
      if ((hourOfDay >= 0 && hourOfDay <= 4) || (hourOfDay >= 10 && hourOfDay <= 15)) {
        record.totalCost = (record.subMetering1 * 4 + record.subMetering2 * 4 + record.subMetering3 * 4) / 1000
      } else if ((hourOfDay >= 5 && hourOfDay <= 6) || (hourOfDay >= 16 && hourOfDay <= 19) || (hourOfDay >= 22 && hourOfDay <= 23)) {
        record.totalCost = (record.subMetering1 * 6 + record.subMetering2 * 6 + record.subMetering3 * 6) / 1000
      } else if (hourOfDay >= 7 && hourOfDay <= 9) {
        record.totalCost = (record.subMetering1 * 12 + record.subMetering2 * 12 + record.subMetering3 * 12) / 1000
      } else {
        record.totalCost = (record.subMetering1 * 10 + record.subMetering2 * 10 + record.subMetering3 * 10) / 1000
      }

      record
    })
  }

Then this hourly data is aggregated into daily basis for which you can refer the code dailyAggregator in the part 2. Get the average cost for a day from this daily aggregated data which represents the revenue loss in the case of outage.

def getRevenueLoss(inputRDD:RDD[Record]):Double = {
    val dataAggregator = new DataAggregator()
    val revenueRDD= getPerHourRevenue(inputRDD)
    val dailyRDD = dataAggregator.dailyAggregator(revenueRDD)
    val totalCostCalcRDD = dailyRDD.map(value => ("sum", value._2.totalCost)).reduceByKey((a, b) => a + b)
    val revenueLossForDayOutage = totalCostCalcRDD.first()._2 / dailyRDD.count()
    revenueLossForDayOutage
  }

So this will end the explanation of the solutions for problem given in hackathon. Each one of us from the team took a particular problem and explained the solution to it. Hope you go through all the solution and get the most out of it.

You can refer here for complete code.

This is the 3rd post in the Predicting Global Energy Demand using Spark series. The series has the following posts.

This is a sequel of the previous blogs from our team members, where the solution for the problem related to energy consumption was explained. In this blog, let me explain our solution to another problem on the same data whose schema is mentioned here.

Problem

What would be household’s peak time load (Peak time is between 7 AM to 10 AM) for the next month.

• During Weekdays?
• During Weekends?

Brief Solution

In order to solve the above stated problem, we need to model the data in such a way that it is aggregated on hourly basis followed by the filtration of peak time records and splitting it into weekday and weekend records. This separated records are aggregated on monthly basis and the peak time load can be calculated on this.

Hourly Aggregation

In order to aggregate the energy consumption data, we need to think about some key factors such as the meter readings add up to form its aggregate whereas voltage and global intensity are averaged out to form its aggregate.

def hourlyAggregator(inputRDD: RDD[Record]): RDD[((String, Long), Record)] = {
    val groupRDD = inputRDD.map(record => ((record.date, record.hourofDay), record)).reduceByKey((firstRecord,
    secondRecord) => {
      val record = new Record()
      record.date = firstRecord.date
      record.day = firstRecord.day
      record.month = firstRecord.month
      record.year = firstRecord.year
      record.hourofDay = firstRecord.hourofDay
      record.subMetering1 = firstRecord.subMetering1 + secondRecord.subMetering1
      record.subMetering2 = firstRecord.subMetering2 + secondRecord.subMetering2
      record.subMetering3 = firstRecord.subMetering3 + secondRecord.subMetering3
      record.activePower = firstRecord.activePower + secondRecord.activePower
      record.reactivePower = firstRecord.reactivePower + secondRecord.reactivePower
      record.voltage = (firstRecord.voltage + secondRecord.voltage) / 2
      record.globalIntensity = (firstRecord.globalIntensity + secondRecord.globalIntensity) / 2
      record
    })
    groupRDD
  }

Monthly Aggregation

Once the hourly aggregation is done, the records are aggregated on daily basis followed by aggregation based on the month.

 def dailyAggregator(inputRDD: RDD[Record]): RDD[(String, Record)] = {
    val groupRDD = inputRDD.map(record => (record.date, record)).reduceByKey((firstRecord, secondRecord) => {
      val record = new Record()
      record.date = firstRecord.date
      record.day = firstRecord.day
      record.month = firstRecord.month
      record.year = firstRecord.year
      record.subMetering1 = firstRecord.subMetering1 + secondRecord.subMetering1
      record.subMetering2 = firstRecord.subMetering2 + secondRecord.subMetering2
      record.subMetering3 = firstRecord.subMetering3 + secondRecord.subMetering3
      record.totalCost = firstRecord.totalCost + secondRecord.totalCost
      record.activePower = firstRecord.activePower + secondRecord.activePower
      record.reactivePower = firstRecord.reactivePower + secondRecord.reactivePower
      record.voltage = (firstRecord.voltage + secondRecord.voltage) / 2
      record.globalIntensity = (firstRecord.globalIntensity + secondRecord.globalIntensity) / 2
      record
    })
    groupRDD
  }

def monthlyAggregator(inputRDD: RDD[Record]): RDD[((Int, Long), Record)] = {
    val groupRDD = inputRDD.map(record => ((record.month, record.year), record)).reduceByKey((firstRecord,
    secondRecord) => {
      val record = new Record()
      record.date = firstRecord.date
      record.day = firstRecord.day
      record.month = firstRecord.month
      record.year = firstRecord.year
      record.subMetering1 = firstRecord.subMetering1 + secondRecord.subMetering1
      record.subMetering2 = firstRecord.subMetering2 + secondRecord.subMetering2
      record.subMetering3 = firstRecord.subMetering3 + secondRecord.subMetering3
      record.totalCost = firstRecord.totalCost + secondRecord.totalCost
      record.activePower = firstRecord.activePower + secondRecord.activePower
      record.reactivePower = firstRecord.reactivePower + secondRecord.reactivePower
      record.voltage = (firstRecord.voltage + secondRecord.voltage) / 2
      record.globalIntensity = (firstRecord.globalIntensity + secondRecord.globalIntensity) / 2
      record
    })
    groupRDD
  }

Calculation of peak time load for the next month

The peak time load for the next month can be predicted using the Geometric Brownian motion algorithm, since the data is aggregated based on the month and this algorithm can be applied upon the data directly

def findNextMonthPeakLoad(rdd:RDD[((Int,Long),Record)],sparkContext:SparkContext) : Double={
      def standardMean(inputRDD: RDD[((Int,Long),Record)]): (List[Double], Double) = {
        val count = inputRDD.count()
        var sum = 0.0
        var riList = List[Double]()
        for (i <- Range(1, count.toInt)) {
          val firstRecord = inputRDD.toArray()(i)
          val secondRecord = inputRDD.toArray()(i - 1)
          val difference = (firstRecord._2.totalPowerUsed -secondRecord._2.totalPowerUsed)
          /firstRecord._2.totalPowerUsed
          riList = riList ::: List(difference)
          sum += difference
        }
        (riList, sum / count)
      }
      def standDeviation(inputRDD:RDD[Double],mean:Double): Double = {
        val sum = inputRDD.map(value => {
          (value - mean) * (value - mean)
        }).reduce((firstValue, secondValue) => {
          firstValue + secondValue
        })
        scala.math.sqrt(sum / inputRDD.count())
      }
      val (rList,mean) = standardMean(rdd)
      val stdDeviation = standDeviation(sparkContext.makeRDD(rList),mean)
      val sortedRdd=rdd.sortByKey(false)
      val lastValue = sortedRdd.first()._2.totalPowerUsed
      var newValues = List[Double]()
      for(i<- Range(0,1000)){
        val predictedValue = lastValue * (1 + mean * 1 + stdDeviation * scala.math.random * 1)
        newValues::=predictedValue
      }
      val sorted = newValues.sorted
      val value = sorted(10)/1000
      value
    }

Hope you understood the idea behind predicting the peak time load for the next month. Further blogs in this series would explain the solution for other problems.

For complete code refer here

This is the 2nd post in the Predicting Global Energy Demand using Spark series. The series has the following posts.

Its our pleasure to share our approach in solving the hackathon problems briefly. So, lets start the exploration with details about the data we handled. The given data represents the energy consumption details recorded for each minute in household usage.The hourly data was having 2075259 measurements gathered between December 2008 and November 2012 (almost 4 years).You can find the schema for data here.

Data Preparation

Data preparation involves pre-processing techniques where we make the data ready for actual prediction. It involves three phases, data cleaning , handling duplicate data and data sampling.

In data cleaning, we handled the missing values in the data.Our approach was to ignore the measurements where there is no readings for that day. Next step is handling duplicates in the data. We ensured there are no duplicates in the data. Final step is data sampling.We drive our process towards perfection by applying it on sampled data. Once we ensure that, the results for the subset holds true. We apply the same process for complete data and publish the results.

Data Modeling

Data modelling involves steps where we model our data in such a way that, it fits into the algorithm used for prediction. We used spark framework and scala language to accomplish this task. In the next sections you will know more about the platform and algorithm we used for prediction.

Scala and Spark

Scala is a JVM based functional language which is well known for its concurrent capabilities.It’s breeze to write map/reduce based machine learning algorithms with very few lines of code. All our examples in the post are written in scala.

Apache Spark is an open source cluster computing system that aims to make data analytics fast, both fast to run and fast to write. We used spark as it allowed us to prototype within time limits of hackathon.

Geometric Brownian Motion

For every prediction problem there will be a strong backbone algorithm which is the key for prediction models. Since the given data is about the energy consumption in household, it is a type of quantity which changes over uncertainty. These kind of quantities follows Geometric Brownian Motion. This model is frequently invoked as a model for diverse quantities.Other quantities which follows this model are stock prices,natural resource prices and growth in demand for products or services etc.

Brownian motion is the simplest of the continuous time stochastic processes.Wiener process gives the mathematical representation for the brownian motion. This model says the variable’s value changes in one unit of time by an amount that is normally distributed with µ and σ ,where µ denotes drift and σ denotes volatility. Brownian motion suggests the following equation to predict the next value at the given instant in constant interval.

Sᵢ˖₁ =SᵢµΔt + Sᵢσԑ√Δt

Where,

• Sᵢ˖₁ : Predicted value
• Sᵢ : Present known value
• µ : Drift
• σ : Volatility
• ԑ : Random Number

Code

µ – Standard Mean ( Drift in data )

standardMean(inputRDD: RDD[(Date, Record)]): (List[Double], Double) = {
val count = inputRDD.count()
var sum = 0.0
var riList = List[Double]()
for (i val firstRecord = inputRDD.toArray()(i)
val secondRecord = inputRDD.toArray()(i - 1)
val difference = (firstRecord._2.totalPowerUsed - secondRecord._2.totalPowerUsed) / firstRecord._2.totalPowerUsed
riList = riList ::: List(difference)
sum += difference
}
(riList, sum / count)
}

σ – Standard Deviation ( Volatility in data )

  def standDeviation(inputRDD: RDD[Double], mean: Double): Double = {
    val sum = inputRDD.map(value => {
      (value - mean) * (value - mean)
    }).reduce((firstValue, secondValue) => {
      firstValue + secondValue
    })
    scala.math.sqrt(sum / inputRDD.count())
  }

Note : Here inputRDD has each measurement aggregated to one day.

Sᵢ˖₁ – Predicted Value

PredictedValue = lastValue * (1 + mean * 1 + stdDeviation * scala.math.random * 1)

Here,

• PredictedValue : Sᵢ˖₁
• lastValue : Sᵢ
• mean : µ
• stdDeviation : σ
• scala.math.random:ԑ

Since we are predicting for next day with the daily aggregated data both Δt and √Δt will be 1.

Hope this blog gave you an idea about our prediction strategy for the next day energy consumption estimation. Wait for the next blog which will explain you the prediction technique for next one year.

This is the first post in the Predicting Global Energy Demand using Spark series. The series has the following posts.

This series of blogs will walk you through our complete solution , designed and implemented for predicting future energy demand using spark.

This problem was solved within 24hrs in hackathon.

Problem statement

Predict the global energy demand for next year using the energy usage data available for last four years, in order to enable utility companies effectively handle the energy demand.

Fluctuating energy demand for utilities is becoming a very big problem. To be more precise about the problem, the demand for energy over a period of time is not consistent. This in turn becomes huge bane for the utilities with varying energy demands. As we know with the existing systems storing of electricity is extremely inefficient, this will increase difficulties for the utilities to bank energy against a time of sudden demand.

As we observe, energy demand rises and falls throughout the day in response to a number of things,including time and environmental factors.The difference between the demand in extremes is important, because utilities must be able to handle demand with supply. One approach to managing electricity demand is building more generation facilities that can be brought online to manage peaks. But using right estimations allows companies to make better decisions.

Available data

Data for energy consumption from 2008-2012 is available here.

The following is the structure of data

• date: Date in format dd/mm/yyyy
• time: time in format hh:mm:ss
• global_active_power: household global minute-averaged active power (in kilowatt)
• global_reactive_power: household global minute-averaged reactive power (in kilowatt)
• Voltage: minute-averaged voltage (in volt)
• global_intensity: household global minute-averaged current intensity (in ampere)
• sub_metering_1: energy sub-metering No. 1 (in watt-hour of active energy). It corresponds to the kitchen, containing mainly a dishwasher, an oven and a microwave (hot plates are not electric but gas powered)
• sub_metering_2: energy sub-metering No. 2 (in watt-hour of active energy). It corresponds to the laundry room, containing a washing-machine, a tumble-drier, a refrigerator and a light.
• Sub_metering_3: energy sub-metering No. 3 (in watt-hour of active energy). It corresponds to an electric water-heater and an air-conditioner

Problems we addressed

• What would be the energy consumption for the next day?
• What would be week wise Energy consumption for the next one year?
• What would be household’s peak time load (Peak time is between 7 AM to 10 AM) for the next month.
  During Weekdays
  During Weekends
• What are the patterns in voltage fluctuations?
• Can you identify the device usage patterns?
• Assuming there was full day of outage, Calculate the Revenue Loss for a particular day next year by finding the Average Revenue per day (ARPD) of the household using given tariff plan

In upcoming blog posts , we will be sharing our solution to above problems.