What Makes Spark Exciting

At Bizo, we’re currently evaluating/prototyping Spark as a replacement for Hive for our batch reports.

As a brief intro, Spark is an alternative to Hadoop. It provides a cluster computing framework for running distributed jobs. Similar to Hadoop, you provide Spark with jobs to run, and it handles splitting up the job into small tasks, assigning those tasks to machines (optionally with Hadoop-style data locality), issuing retries if tasks fail transiently, etc.

In our case, these jobs are processing a non-trivial amount of data (log files) on a regular basis, for which we currently use Hive.

Why Replace Hive?

Admittedly, Hive has served us well for quite awhile now. (One of our engineers even built a custom “Hadoop on demand” framework for running periodic on-demand Hadoop/Hive jobs in EC2 several months before Amazon Elastic Map Reduce came out.)

Without Hive, it would have been hard for us to provide the same functionality, probably at all, let alone in the same time frame.

That said, it has gotten to the point where Hive is more frequently invoked in negative contexts (“damn it, Hive”) than positive.

Personally, I admittedly even try to avoid tasks that involve working with Hive. I find it to be frustrating and, well, just not a lot of fun. Why? Two primary reasons:

1. Hive jobs are hard to test

Bizo has a culture of excellence, and for engineering one of the things this means is testing. We really like tests. Especially unit tests, which are quick to run and enable a fast TDD cycle.

Unfortunately, Hive makes unit testing basically impossible. For several reasons:

• Hive scripts must be run in a local Hadoop/Hive installation.

  Ironically, very few developers at Bizo have local Hadoop installations. We are admittedly spoiled by Elastic Map Reduce, such that most of us (myself anyway) wouldn’t even know how to setup Hadoop off the top of our heads. We just fire up an EMR cluster.

• Hive scripts have production locations embedded in them.

  Both our log files and report output are stored in S3, so our Hive scripts end up with lots of “s3://” paths scattered throughout in them.

  While we do run dev versions of reports with “-dev” S3 buckets, still relying on S3 and raw log files (that are usually in a compressed/binary-ish format) is not conducive to setting up lots of really small, simplified scenarios to unit test each boundary case.

• Hive scripts do not provide any abstraction–they are just one big HiveQL file. This means its hard to break up a large report into small, individually testable steps.

Despite these limitations, about a year ago we had a developer dedicate some effort to prototyping an approach that would run Hive scripts within our CI workflow. In the end, while his prototype worked, the workflow was wonky enough that we never adopted it for production projects.

The result? Our Hive reports are basically untested. This sucks.

2. Hive is hard to extend

Extending Hive via custom functions (UDFs and UDAFs) is possible, and we do it all the time–but it’s a pain in the ass.

Perhaps this is not Hive’s fault, and it’s some Hadoop internals leaking into Hive, but the various ObjectInspector hoops, to me, always seemed annoying to deal with.

Given these shortcomings, Bizo has been looking for a Hive-successor for awhile, even going so far as to prototype revolute, a Scala DSL on top of Cascading, but had not yet found something we were really excited about.

Enter Spark!

We had heard about Spark, but did not start trying it until being so impressed by the Spark presentation at AWS re:Invent (the talk received the highest rating of all non-keynote sessions) that we wanted to learn more.

One of Spark’s touted strengths is being able to load and keep data in memory, so your queries aren’t always I/O bound.

That is great, but the exciting aspect for us at Bizo is how Spark, either intentionally or serendipitously, addresses both of Hive’s primary shortcomings, and turns them into huge strengths. Specifically:

1. Spark jobs are amazingly easy to test

Writing a test in Spark is as easy as:

class SparkTest {
  @Test
  def test() {
    // this is real code...
    val sc = new SparkContext("local", "MyUnitTest')
    // and now some psuedo code...
    val output = runYourCodeThatUsesSpark(sc)
    assertAgainst(output)
  }
}

(I will go into more detail about runYourCodeThatUsesSpark in a future post.)

This one liner starts up a new SparkContext, which is all your program needs to execute Spark jobs. There is no local installation required (just have the Spark jar on your classpath, e.g. via Maven or Ivy), no local server to start/stop. It just works.

As a technical aside, this “local” mode starts up an in-process Spark instance, backed by a thread-pool, and actually opens up a few ports and temp directories, because it’s a real, live Spark instance.

Granted, this is usually more work than you want to be done in an unit test (which ideally would not hit any file or network I/O), but the redeeming quality is that it’s fast. Tests run in ~2 seconds.

Okay, yes, this is slow compared to pure, traditional unit tests, but is such a huge revolution compared to Hive that we’ll gladly take it.

2. Spark is easy to extend

Spark’s primary API is a Scala DSL, oriented around what they call an RDD, or Resilient Distributed Dataset, which is basically a collection that only supports bulk/aggregate transforms (so methods like map, filter, and groupBy, which can be seen as transforming the entire collection, but no methods like get or take which assume in-memory/random access).

Some really short, made up example code is:

// RDD[String] is like a collection of lines
val in: RDD[String] = sc.textFile("s3://bucket/path/")
// perform some operation on each line
val suffixed = in.map { line => line + "some suffix" }
// now save the new lines back out
suffixed.saveAsTextFile("s3://bucket/path2")

Spark’s job is to package up your map closure, and run it against that extra large text file across your cluster. And it does so by, after shuffling the code and data around, actually calling your closure (i.e. there is no LINQ-like introspection of the closure’s AST).

This may seem minor, but it’s huge, because it means there is no framework code or APIs standing between your running closure and any custom functions you’d want to run. Let’s say you want to use SomeUtilityClass (or the venerable StringUtils), just do:

import com.company.SomeUtilityClass
val in: RDD[String] = sc.textFile("s3://bucket/path/")
val processed = in.map { line =>
  // just call it, it's a normal method call
  SomeUtilityClass.process(line) 
}
processed.saveAsTextFile("s3://bucket/path2")

Notice how SomeUtilityClass doesn’t have to know it’s running within a Spark RDD in the cluster. It just takes a String. Done.

Similarly, Spark doesn’t need to know anything about the code you use witin the closure, it just needs to be available on the classpath of each machine in the cluster (which is easy to do as part of your cluster/job setup, you just copy some jars around).

This seamless hop between the RDD and custom Java/Scala code is very nice, and means your Spark jobs end up reading just like regular, normal Scala code (which to us is a good thing!).

Is Spark Perfect?

As full disclosure, we’re still in the early stages of testing Spark, so we can’t yet say whether Spark will be a wholesale replacement for Hive within Bizo. We haven’t gotten to any serious performance comparisons or written large, complex reports to see if Spark can take whatever we throw at it.

Personally, I am also admittedly somewhat infutuated with Spark at this point, so that could be clouding my judgement about the pros/cons and the tradeoffs with Hive.

One Spark con so far is that Spark is pre-1.0, and it can show. I’ve seen some stack traces that shouldn’t happen, and some usability warts, that hopefully will be cleared up by 1.0. (That said, even as a newbie I find the codebase small and very easy to read, such that I’ve had several small pull requests accepted already–which is a nice consolation compared to the daunting codebases of Hadoop and Hive.)

We have also seen that, for our first Spark job, moving from “Spark job written” to “Spark job running in production” is taking longer than expected. But given that Spark is a new tool to us, we expect this to be a one-time cost.

More to Come

I have a few more posts coming up which explain our approach to Spark in more detail, for example:

• Testing best practices
• Running Spark in EMR
• Accessing partitioned S3 logs

To see those when they come out, make sure to subscribe to the blog, or, better yet, come work at Bizo and help us out!

Grouping pageviews into visits: a Scala code kata

The basic units of any website traffic analysis are pageviews, visits, and unique visitors.  Tracking pageviews is simply a matter of counting requests to the server.  Calculating unique visitors usually relies on cookies and unique identifiers.  Visits, however, require a bit more work.  For our purposes, a single visit is defined as a sequence of pageviews where the interval between pageviews is less than a fixed length like 15 minutes.

I thought that the problem of grouping pageviews into visits would make an interesting code kata.  Here’s the statement of the problem that I worked from:

Given a non-empty sequence of timestamps (as milliseconds since the epoch), write a function that would return a sequence of visits, where each visit is itself a sequence of timestamps where each pair of consecutive timestamps is no more than N milliseconds apart.

As a starting point, I decided to take a straightforward procedural approach:

def doingItIteratively(pageviews: Seq[Long]): Seq[Seq[Long]] = {
 val iterator = pageviews.sorted.iterator
 val visits = ListBuffer[ListBuffer[Long]]()

 var previousPV: Long = iterator.next
 var currentVisit: ListBuffer[Long] = ListBuffer(previousPV)

 for (currentPV <- iterator) {
   if (currentPV - previousPV > N) {
     visits += currentVisit
     currentVisit = ListBuffer[Long]()
   }

   currentVisit += currentPV
   previousPV = currentPV
 }
 visits += currentVisit

 visits map (_.toSeq) toSeq
}

So, we simply iterate through the (sorted) events tracking both the current visit and the previous pageview.  If the current pageview represents a new visit, push the previous visit into the list of all visits and start a new one.  Then push the current pageview into the (potentially new) visit.

It actually felt a bit odd to write procedural code like this and ignore the functional parts of Scala.  Using a fold cleans the code up a bit and gets rid of the mutable state.

def doingItByFolds(pageviews: Seq[Long]): Seq[Seq[Long]] = {
 val sortedPVs = pageviews.sorted

 (Seq[Seq[Long]]() /: sortedPVs) { (visits, pv) =>
   val isNewVisit = visits.lastOption flatMap (_.lastOption) map {
     prevPV => pv - prevPV > N
   } getOrElse true

   if (isNewVisit) {
     visits :+ Seq(pv)
   } else {
     visits.init :+ (visits.last :+ pv)
   }
 }
}

Here, we’re starting with an empty list of visits and folding it over the sorted pageviews.  At each pageview, we decide if we need to start a new visit.  If so, we append a new visit containing the pageview to the accumulated visits.  If not, we pop off the last visit, append the pageview, and put the last visit back on the tail of the accumulated visits.

One part that’s still a bit messy is comparing the current timestamp to the previous one.  We can improve that by iterating through the intervals between pageviews instead of the actual pageviews.

def slidingThroughIt(pageviews: Seq[Long]): Seq[Seq[Long]] = {
 val intervals = (0L +: pageviews.sorted).sliding(2)

 (Seq[Seq[Long]]() /: intervals) {
   (visits, interval) =>
     if (interval(1) - interval(0) > N) {
       visits :+ Seq(interval(1))
     } else {
       visits.init :+ (visits.last :+ interval(1))
     }
 }
}

Here, we’re prepending a “0L” timestamp (and assuming that none of the pageviews happened in the early 70s) and using the “sliding” method to pair each timestamp with the previous one.

So far, we’ve been using a sequence of pageviews as a visit.  What happens if we add an explicit Visit type?  This lets us convert all pageviews into Visits at the start, then focus on merging overlapping Visits.  One nice benefit is that this is a map-reduce algorithm that can be easily parallelized instead of one that must sequentially iterate over the pageviews (either explicitly or with a fold).

case class Visit(start: Long, end: Long, pageviews: Seq[Long]) {
 def +(other: Visit): Visit = {
   Visit(min(start,other.start), max(end, other.end),
         (pageviews ++ other.pageviews).sorted)
 }
}

def doingItMapReduceStyle(pageviews: Seq[Long]): Seq[Visit] = {
 pageviews.par map { pv =>
   Seq(Visit(pv, pv+N, Seq(pv))
 } reduce { (visit1, visit2) =>
   val sortedVisits = (v1 ++ v2) sortBy (_.start)

   (Seq[Visit]() /: sortedVisits) { (visits, next) =>
     if (visits.lastOption map(_.end >= next.start) getOrElse false)
     {
       visits.init :+ (visits.last + visit)
     } else {
       visits :+ visit
     }
   }
 }
}

The map-reduce solution is fun, but in a production system, I’d probably stick with the sliding variation and add a bit more flexibility to track actual pageview objects instead of just timestamps.

Using GROUP BYs or multiple INSERTs with complex data types in Hive.

In any sort of ad hoc data analysis, the first step is often to extract a specific subset of log lines from our files.  For example, when looking at a single partner’s web traffic, I often use an initial query to copy that partner’s data into a new table.  In addition to segregating out only the data relevant to my analysis, I use this to copy the data from S3 into HDFS, which will make later queries more efficient.  (Using maps as our log lines is how we support dynamic columns.)

create external table if not exists
original_logs(fields map<string,string>) location “...” ;

create table if not exists
extracted_logs(fields map<string,string>) ;

insert overwrite table extracted_logs
select * from original_logs where fields[“partnerId”] = 123 ;

If I’m doing this for multiple partners, it’s tempting to use a multiple-insert so Hadoop only needs to make one pass of the original data.

create external table if not exists
original_logs(fields map<string,string>) location “...” ;

create table if not exists
extracted_logs(fields map<string,string>)
partitioned by (partnerId int);

from original_logs
insert overwrite table extracted_logs partition (partnerId = 123)
select * from original_logs where fields[“partnerId”] = 123
insert overwrite table extracted_logs partition (partnerId = 234)
select * from original_logs where fields[“partnerId”] = 234

Unfortunately, in Hive 0.7.x, this query fails with the error message “Hash code on complex types not supported yet.”  A multiple-insert statement uses an implicit group by, and Hive 0.7.x does not support grouping by complex types.  This bug was partially addressed in 0.8, which added support for arrays and maps, but structs and unions are still not supported.

At an initial glance, it does look like adding this support should be straightforward.  This could be a good candidate for our next open source day.

mdadm: device or resource busy

I just spent a few hours tracking an issue with mdadm (Linux utility used to manage software RAID devices) and figured I'd write a quick blog post to share the solution so others don't have to waste time on the same.

As a short background, we use mdadm to create RAID-0 stripped devices for our Sugarcube analytics (OLAP) servers using Amazon EBS volumes.

The issue manifested itself as a random failure during device creation:


$ mdadm --create /dev/md0 --level=0 --chunk 256 --raid-devices=4 /dev/xvdh1 /dev/xvdh2 /dev/xvdh3 /dev/xvdh4
mdadm: Defaulting to version 1.2 metadata
mdadm: ADD_NEW_DISK for /dev/xvdh3 failed: Device or resource busy

I searched and searched the interwebs and tried every trick I found to no avail. We don't have dmraid installed on our Linux images (Ubuntu 12.04 LTS / Alestic cloud image) so there's no possible conflict there.  All devices were clean, as they are freshly created EBS volumes and I knew none of them were in use.  

Before running mdadm --create, mdstat was clean:

$ cat /proc/mdstat

Personalities : [linear] [multipath] [raid0] [raid1] [raid6] [raid5] [raid4] [raid10]

unused devices: <none>

And yet after running it the devices were assigned to two different devices instead of just /dev/md0:

$ cat /proc/mdstat

Personalities : [linear] [multipath] [raid0] [raid1] [raid6] [raid5] [raid4] [raid10]

md127 : inactive xvdh4[3](S) xvdh3[2](S)

      1048573952 blocks super 1.2

md0 : inactive xvdh2[1](S) xvdh1[0](S)

      1048573952 blocks super 1.2

unused devices: <none>

Looking into dmesg didn't reveal anything interesting either:

$ dmesg 

...

[3963010.552493] md: bind<xvdh1>

[3963010.553011] md: bind<xvdh2>

[3963010.553040] md: could not open unknown-block(202,115).

[3963010.553052] md: md_import_device returned -16

[3963010.566543] md: bind<xvdh3>

[3963010.731009] md: bind<xvdh4>

And strangely, the creation or assembly would sometime work and sometime not:

$ mdadm --manage /dev/md0 --stop

mdadm: stopped /dev/md0

$ sudo mdadm --assemble --force /dev/md0 /dev/xvdh[1234]

mdadm: /dev/md0 has been started with 4 drives.

$ mdadm --manage /dev/md0 --stop

mdadm: stopped /dev/md0

$ sudo mdadm --assemble --force /dev/md0 /dev/xvdh[1234]

mdadm: cannot open device /dev/xvdh3: Device or resource busy

$ mdadm --manage /dev/md0 --stop

mdadm: stopped /dev/md0

$ sudo mdadm --assemble --force /dev/md0 /dev/xvdh[1234]

mdadm: cannot open device /dev/xvdh1: Device or resource busy

mdadm: /dev/xvdh1 has no superblock - assembly aborted

$ mdadm --manage /dev/md0 --stop

mdadm: stopped /dev/md0

$ sudo mdadm --assemble --force /dev/md0 /dev/xvdh[1234]

mdadm: /dev/md0 has been started with 4 drives.

I started suspecting I was facing some kind of underlying race condition where the devices would get assigned/locked during the device creation process.   So I started googling for "mdadm create race" and  I finally found a post that tipped me off. While it didn't provide the solution, the post put me on the right track by mentioning udev and it took only a few more minutes to narrow down on the solution:  disabling udev events during device creation to avoid contention on device handles.

So now our script goes something like:

$ udevadm control --stop-exec-queue

$ mdadm --create /dev/md0 --run --level=0 --raid-devices=4 ...

$ udevadm control --start-exec-queue

And we now have consistent reliable device creation.

Hopefully this blog post will help other passers-by with a similar problem.  Good luck!

Amazon Web Services Outages: 4 Steps for Survival

(Cross-post from the Bizo Blog)

Another Amazon Web Services (AWS) cloud outage over the past weekend took down some pretty major services such as Netflix, Heroku, Pinterest and Instagram.  At Bizo, a company that provides business marketing services for hundreds of F1000 clients, we serve billions of requests a day across tens of thousands of websites, and have our entire infrastructure on the AWS cloud, but didn’t have any downtime.  The simple reason is that we take our customers’ uptime and site performance seriously, and have built tools and services on AWS to ensure high-availability (HA) and low-latency (LL) services.  Despite the FUD created by many of the industry blogs and press, it is possible to create HA and LL services on AWS if you follow some simple steps.

Amazon Web Services Outages: 4 Steps for Survival

AWS Billing Info in Hive

Amazon recently (finally!) launched programmatic access to your AWS billing data.

Once you turn it on, select a bucket, grant access to the AWS system user, you'll get a .csv file with your estimated billing for the month. The files are delivered daily, but they contain month-to-date information, and will replace the file from the previous day.

It's easy enough to view this information in excel (or similar), but I thought it would be fun to take a look in hive, especially once we start having data for a few months to aggregate over.

Amazon delivers the data to the root of your bucket. I decided to start moving it to a hive-partitioned path, to make it easier to query once we start have more data. I wrote a simple scala script to move the data to [bucket]/partioned/year=[year]/month=[month]/[file]. Here's some example code.

Ok, now we're ready to read the data in Hive.

Here's a hive schema for the AWS billing information. It uses the csv-serde (make sure you add that jar before running the create table statement). Run alter table aws_billing recover partitions; to load in the partitions (one per year/month), and you're ready to query.

Like I said, it's overkill to use hive to read this data for a month or so, but it's just so addictive having a SQL interface to arbitrary S3 data :).

Here are some example queries to get you started.

Costs by Service

select ProductCode, UsageType, Operation, sum(TotalCost)
  from aws_billing
 where RecordType in ("PayerLineItem", "LinkedLineItem")
 group
    by ProductCode, UsageType, Operation
;
  

EC2 usage, by size (across EC2/EMR)

select ProductCode, UsageType,sum(TotalCost)
  from aws_billing
 where RecordType in ("PayerLineItem", "LinkedLineItem")
   and UsageType like "BoxUsage%"
 group
    by ProductCode, UsageType
;
  

the golden rule of programming style

There's an interesting page on the subject of compilation units per file over at the scala style guide.

The guideline, is, delightfully vague, which I will paraphrase as: Mostly use single files, unless you can't, or unless it's better if you don't.

The author(s) go on to expand on the reasoning behind breaking the guideline:

Another case is when multiple classes logically form a single, cohesive group, sharing concepts to the point where maintenance is greatly served by containing them within a single file. These situations are harder to predict… Generally speaking, if it is easier to perform long-term maintenance and development on several units in a single file rather than spread across multiple, then such an organizational strategy should be preferred for these classes.

This touches on what I consider to be the golden rule of programming style: Make your intent clear and the code easy to read.

Software spends most of its life in maintenance, which is why we have style guides and coding standards. It's valuable to have consistent looking code to promote a shared vocabulary, improve readability, and steer away from confusing or error-prone constructs.

It is just as important to be able to understand, both as an author and as a reviewer, that in certain cases following the letter of the law goes against the main goal of improving readability and maintenance. A one-size-fits-all rule does not always work, and as the authors of this particular guideline mention, "these situations are harder to predict."

Make your intent clear and the code easy to read.