Scala and Data Pipeline - Phone Bill Calculation Example

Earlier this year a friend sent me a video showing how he implemented a phone bill calculation challenge using Scala. I took a stab at it using Java + Data Pipeline and below is what I came up with.

How about you? How would you code this using your favourite language or framework?

The Scala Approach

You can watch Ervin’s walk-through of his Scala solution:

http://www.hardcomsoft.com/videos/scala_sample_billCalculation.html

You can also read the original requirements for yourself.

The Data Pipeline Approach

Here’s the Data Pipeline code I put together.

It uses a CSVReader to load the data from a string.
It then uses a series of TransformingReaders to convert the duration to seconds and conditionally calculate the price depending on whether the duration is over 5 minutes or not.
Next it uses a GroupByReader to sum the price and duration and to count the calls before sorting them and dropping the one with the highest duration.
Finally, another GroupByReader is used to sum all the prices together to find and return the grand total.

import java.io.StringReader;

import com.northconcepts.datapipeline.core.DataReader;
import com.northconcepts.datapipeline.core.LimitReader;
import com.northconcepts.datapipeline.core.SortingReader;
import com.northconcepts.datapipeline.csv.CSVReader;
import com.northconcepts.datapipeline.filter.FilterExpression;
import com.northconcepts.datapipeline.group.GroupByReader;
import com.northconcepts.datapipeline.job.Job;
import com.northconcepts.datapipeline.memory.MemoryWriter;
import com.northconcepts.datapipeline.transform.RenameField;
import com.northconcepts.datapipeline.transform.SetCalculatedField;
import com.northconcepts.datapipeline.transform.SetField;
import com.northconcepts.datapipeline.transform.TransformingReader;

public class ScratchPhoneCharges {

    private static String INPUT = "00:01:07,400-234-090\n" + 
                                  "00:05:01,701-080-080\n" + 
                                  "00:05:00,400-234-090";

    private static final int SECOND_FEE = 3;
    private static final int MINUTE_FEE = 150;
    private static final int FEE_CHANGE_THRESHOLD_IN_MIN = 5;
    
    public static void main(String[] args) {
        int priceInCents = solution(INPUT);
        System.out.println("price in cents = " + priceInCents);
    }
    
    public static int solution(String s) {
        DataReader reader = new CSVReader(new StringReader(s));
        
        reader = new TransformingReader(reader)
                .add(new RenameField("A", "duration"))
                .add(new RenameField("B", "phone"))
                .add(new SetCalculatedField("duration_seconds", 
                        "(toInt(substring(duration, 0, 2)) * 3600) + " +
                        "(toInt(substring(duration, 3, 5)) * 60) + " +
                        "toInt(substring(duration, 6, 8))"));

        // if less than 5 minutes
        reader = new TransformingReader(reader)
                .setCondition(new FilterExpression("duration_seconds < " + FEE_CHANGE_THRESHOLD_IN_MIN + " * 60"))
                .add(new SetField("rate_units", "per second")) 
                .add(new SetField("rate", SECOND_FEE)) 
                .add(new SetCalculatedField("price", "duration_seconds * rate"));

        // if 5 minutes or more
        reader = new TransformingReader(reader)
                .setCondition(new FilterExpression("duration_seconds >= " + FEE_CHANGE_THRESHOLD_IN_MIN + " * 60"))
                .add(new SetField("rate_units", "per minute")) 
                .add(new SetField("rate", MINUTE_FEE)) 
                .add(new SetCalculatedField("price", "(duration_seconds+59)/60 * rate"));
        
        reader = new GroupByReader(reader, "phone").sum("price").sum("duration_seconds").count("calls");
        
        reader = new SortingReader(reader).desc("duration_seconds").asc("phone");
        
        // skip first record (phone with highest duration)
        reader = new LimitReader(reader).setOffset(1);
        
        reader = new GroupByReader(reader).sum("price").sum("calls");
        
        MemoryWriter writer = new MemoryWriter();
        
        Job.run(reader, writer);
        
        return writer.getRecordList().get(0).getField("price").getValueAsInteger();
    }

}

import java.io.StringReader;

import com.northconcepts.datapipeline.core.DataReader;

import com.northconcepts.datapipeline.core.LimitReader;

import com.northconcepts.datapipeline.core.SortingReader;

import com.northconcepts.datapipeline.csv.CSVReader;

import com.northconcepts.datapipeline.filter.FilterExpression;

import com.northconcepts.datapipeline.group.GroupByReader;

import com.northconcepts.datapipeline.job.Job;

import com.northconcepts.datapipeline.memory.MemoryWriter;

import com.northconcepts.datapipeline.transform.RenameField;

import com.northconcepts.datapipeline.transform.SetCalculatedField;

import com.northconcepts.datapipeline.transform.SetField;

import com.northconcepts.datapipeline.transform.TransformingReader;

public class ScratchPhoneCharges {

private static String INPUT = "00:01:07,400-234-090\n" +

"00:05:01,701-080-080\n" +

"00:05:00,400-234-090";

private static final int SECOND_FEE = 3;

private static final int MINUTE_FEE = 150;

private static final int FEE_CHANGE_THRESHOLD_IN_MIN = 5;

public static void main(String[] args) {

int priceInCents = solution(INPUT);

System.out.println("price in cents = " + priceInCents);

}

public static int solution(String s) {

DataReader reader = new CSVReader(new StringReader(s));

reader = new TransformingReader(reader)

.add(new RenameField("A", "duration"))

.add(new RenameField("B", "phone"))

.add(new SetCalculatedField("duration_seconds",

"(toInt(substring(duration, 0, 2)) * 3600) + " +

"(toInt(substring(duration, 3, 5)) * 60) + " +

"toInt(substring(duration, 6, 8))"));

// if less than 5 minutes

reader = new TransformingReader(reader)

.setCondition(new FilterExpression("duration_seconds < " + FEE_CHANGE_THRESHOLD_IN_MIN + " * 60"))

.add(new SetField("rate_units", "per second"))

.add(new SetField("rate", SECOND_FEE))

.add(new SetCalculatedField("price", "duration_seconds * rate"));

// if 5 minutes or more

reader = new TransformingReader(reader)

.setCondition(new FilterExpression("duration_seconds >= " + FEE_CHANGE_THRESHOLD_IN_MIN + " * 60"))

.add(new SetField("rate_units", "per minute"))

.add(new SetField("rate", MINUTE_FEE))

.add(new SetCalculatedField("price", "(duration_seconds+59)/60 * rate"));

reader = new GroupByReader(reader, "phone").sum("price").sum("duration_seconds").count("calls");

reader = new SortingReader(reader).desc("duration_seconds").asc("phone");

// skip first record (phone with highest duration)

reader = new LimitReader(reader).setOffset(1);

reader = new GroupByReader(reader).sum("price").sum("calls");

MemoryWriter writer = new MemoryWriter();

Job.run(reader, writer);

return writer.getRecordList().get(0).getField("price").getValueAsInteger();

}

What’s your Approach?

I’d love to see other approaches in any programming language or framework, but other Scala and Java approaches would be great to see.

Happy Coding!

Scala and Data Pipeline – Phone Bill Calculation Example

The Scala Approach

The Data Pipeline Approach

What’s your Approach?

About Dele Taylor

Data Pipeline

Docs

Company

Tools