Wednesday, September 29, 2010

emr: Cannot run program "bash": java.io.IOException: error=12, Cannot allocate memory

Moving one of our jobs from hive 0.4 / hadoop 0.18 to hive 0.5 / hadoop 0.20 on amazon emr, I ran into a weird error in the reduce stage, something like:


java.io.IOException: Task: attempt_201007141555_0001_r_000009_0 - The reduce copier failed
at org.apache.hadoop.mapred.ReduceTask.run(ReduceTask.java:384)
at org.apache.hadoop.mapred.Child.main(Child.java:170)
Caused by: java.io.IOException: Cannot run program "bash": java.io.IOException: error=12, Cannot allocate memory
at java.lang.ProcessBuilder.start(ProcessBuilder.java:459)
at org.apache.hadoop.util.Shell.runCommand(Shell.java:149)
at org.apache.hadoop.util.Shell.run(Shell.java:134)
at org.apache.hadoop.fs.DF.getAvailable(DF.java:73)
at org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.getLocalPathForWrite(LocalDirAllocator.java:329)
at org.apache.hadoop.fs.LocalDirAllocator.getLocalPathForWrite(LocalDirAllocator.java:124)
at org.apache.hadoop.mapred.MapOutputFile.getInputFileForWrite(MapOutputFile.java:160)
at org.apache.hadoop.mapred.ReduceTask$ReduceCopier$InMemFSMergeThread.doInMemMerge(ReduceTask.java:2622)
at org.apache.hadoop.mapred.ReduceTask$ReduceCopier$InMemFSMergeThread.run(ReduceTask.java:2586)
Caused by: java.io.IOException: java.io.IOException: error=12, Cannot allocate memory
at java.lang.UNIXProcess.(UNIXProcess.java:148)
at java.lang.ProcessImpl.start(ProcessImpl.java:65)
at java.lang.ProcessBuilder.start(ProcessBuilder.java:452)
... 8 more


There's some discussion on this thread in the emr forums.

From Andrew's response to the thread:


The issue here is that when Java tries to fork a process (in this case bash), Linux allocates as much memory as the current Java process, even though the command you are running might use very little memory. When you have a large process on a machine that is low on memory this fork can fail because it is unable to allocate that memory.


The workaround here is to either use an instance with more memory (m2 class), or reduce the number of mappers or reducers you are running on each machine to free up some memory.

Since the task I was running was reduce heavy, I chose to just drop the number of mappers from 4 to 2. You can do this pretty easy with the emr bootstrap actions.

My job ended up looking something like this:


elastic-mapreduce --create --name "awesome script" \
--num-instances 8 --instance-type m1.large \
--hadoop-version 0.20 \
--bootstrap-action s3://elasticmapreduce/bootstrap-actions/configure-hadoop \
--args "-s,mapred.tasktracker.map.tasks.maximum=2" \
--hive-script --arg s3://....../script

(relevant parts highlighted).


Tuesday, September 21, 2010

Salesforce and DART Synchronization

I’ve recently started some work that involves extending Salesforce for our Ad Ops team. For our most recent Hack Day, I decided to do a little project to continue learning about development with the Salesforce cloud platform, Force.com.

After thinking about what I wanted to work on, I decided to build a custom button that would allow a user to update an Account record in Salesforce with an Advertiser ID from DART, our primary ad serving platform, for the following reasons:
  1. It’s a tool that I could see being used in our live Salesforce instance.
  2. It seems like a typical use case for extending Salesforce (i.e. integrating with a 3rd party SOAP service).
The back of the napkin design looked like this:



At a high-level, I wanted to call DART’s DFP API from within Salesforce and then update an Account object in Salesforce with the Advertiser Id returned from DART. However, I first needed to authenticate with Google’s ClientLogin service in order to get an authentication token for calling the DFP API.

APEX

APEX is the programming language that allows a developer to customize a Salesforce installation. APEX’s syntax, not surprisingly, is very similar to Java. The really interesting thing is that none of the code you write actually compiles or runs on your machine. All compilation and execution happen “in the cloud”.

DART Integration

Salesforce has a strict security model. In order to make a request to a Web Service you actually need to configure any URLs you are accessing as a Remote Site. Instructions for doing this can be found here. For this project, I simply needed to add https://www.google.com as a Remote Site.

There are a couple of options for calling a Web Service via APEX:
  • Use the Http/HttpRequest/Http APEX classes. These are useful for calling REST style services.
  • Import a WSDL and use the generated code to make a SOAP request.
In this project, I ended up using both methods.

Here is the APEX code I developed for calling Google’s ClientLogin authentication service:
public class GoogleAuthIntegration {
private static String CLIENT_AUTH_URL = 'https://www.google.com/accounts/ClientLogin';

// login to google with the given email and password
public static String performClientLogin(final String email, final String password) {
final Http http = new Http();
final HttpRequest request = new HttpRequest();
request.setEndpoint(CLIENT_AUTH_URL);
request.setMethod('POST');
request.setHeader('Content-type', 'application/x-www-form-urlencoded');

final String body = 'service=gam&accountType=GOOGLE&' + 'Email=' + email + '&Passwd=' + password;
request.setBody(body);

final HttpResponse response = http.send(request);
final String responseBody = response.getBody();
final String authToken = responseBody.substring(responseBody.indexOf('Auth=') + 5).trim();

System.debug('authToken is: ' + authToken);
return authToken;
}
}
This piece of code would fetch an authToken for the given username and password. Once I had the authToken, I could then call the DFP API. For this part, I used WSDL/SOAP, the 2nd method for calling web services.

Salesforce provides a way to import a WSDL file via its Admin UI. It then parses and generates APEX code that allows you to call methods exposed by the WSDL. However, when I tried importing DFP’s Company Service WSDL, I ran into some errors:



It turns out that the WSDL contains an element named ‘trigger’ and trigger is a reserved APEX keyword. In any event, I ended up copy/pasting the generated code and fixing it so that it compiled correctly (I also ran into a problem where generated exception classes were not extending Exception).

Once the code to call the DFP Company Service was compiling, I created an APEX controller to perform the update on an Account record.
public class SyncDartAccountController {
private final Account acct;

public SyncDartAccountController(ApexPages.StandardController stdController) {
this.acct = (Account) stdController.getRecord();
}

// Code we will invoke on page load.
public PageReference onLoad() {
String theId = ApexPages.currentPage().getParameters().get('id');

if (theId == null) {
// Display the Visualforce page's content if no Id is passed over
return null;
}

// get authToken for DFP API requests
String authToken = GoogleAuthIntegration.performClientLogin('xxx@xxx.com', 'xxxx');

// get Account with the given id
for (Account o:[select id, name from Account where id =:theId]) {
DartCompanyService.CompanyServiceInterfacePort p = new DartCompanyService.CompanyServiceInterfacePort();
p.RequestHeader = new DartCompanyService.SoapRequestHeader();
p.RequestHeader.applicationName = 'sampleapp';

// prepare the DFP query and execute
DartCompanyService.Statement filterByNameAndType = new DartCompanyService.Statement();
filterByNameAndType.query = 'WHERE name = \'' + o.Name + '\' and type = \'ADVERTISER\'';

DartCompanyService.CompanyPage page = p.getCompaniesByStatement(filterByNameAndType);

if (page.totalResultSetSize > 0) {
// update the record if we get a result
o.Dart_Advertiser_Id__c = page.results.get(0).id;
update o;
}
}

// Redirect the user back to the original page
PageReference pageRef = new PageReference('/' + theId);
pageRef.setRedirect(true);
return pageRef;
}
}
UI updates

Then, I created a simple Visuaforce page to invoke the controller:

<apex:page standardController="Account" extensions="SyncDartAccountController" action="{!onLoad}">
<apex:sectionHeader title="Auto-Running Apex Code"/>
<apex:outputPanel >
You tried calling Apex Code from a button. If you see this page, something went wrong.
You should have been redirected back to the record you clicked the button from.
</apex:outputPanel>
</apex:page>

Finally, I added a custom button to the Account page which would invoke the Visualforce page. You can do this in the Salesforce UI:

1) Click on ‘Buttons and Links’:



2) Click New:


3) Enter the info for the new button:



4) After clicking on Save, we can add the button to the Account page layout. The final result:

Final Thoughts
This was my first foray into APEX programming in Salesforce and I was pleased with the overall set of tools and ability to be productive quickly. The only hiccup I encoutered was in the WSDL generation step and this issue was fairly easy to overcome. There are good developer docs and there are ways to add debug logging (which I didn’t go over) as well as a framework for unit testing.

Monday, September 20, 2010

quick script: emr-mailer

We write a lot of hive reports. Frequently we want to email the resulting report to a list. In the past I've usually done this with some one-off post processing scripts, but I thought it would be nice to write a reusable emr job step that will execute as part of the hive job.

The script will download files from an s3 url, concatenate them together, zip up the results and send it as an attachment to a specified email address. It sends email through smtp.mail.com, using account credentials you specify.

I wanted to make it easy to just append an additional step to any existing job, not requiring any additional machine setup or dependencies. I was able do this by making use of amazon's script-runner (s3://us-east-1.elasticmapreduce/libs/script-runner/script-runner.jar). The script-runner.jar step will let you execute an arbitrary script from a location in s3 as an emr job step.

As I mentioned, the intended usage is to run it as a job step with your hive script, passing it in the location of the resulting report.

E.g.:


elastic-mapreduce --create --name "my awesome report ${MONTH}" \
--num-instances 10 --instance-type c1.medium --hadoop-version 0.20 \
--hive-script --arg s3://path/to/hive/script.sql \
--args -d,MONTH=${MONTH} --args -d,START=${START} --args -d,END=${END} \
--jar s3://us-east-1.elasticmapreduce/libs/script-runner/script-runner.jar \
--args s3://path/to/emr-mailer/send-report.rb \
--args -n,report_${MONTH} --args -s,"my awesome report ${MONTH}" \
--args -e,awesome-reports@company.com \
--args -r,s3://path/to/report/results


Above you can see I'm starting a hive report as normal, then simply appending the script-runner step, calling the emr-mailer send-report.rb, telling it where the report will end up, and details about the email.


The full source code is available on github as emr-mailer.

The script is pretty simple, but let me know if you have any suggestions for improvements or other feedback.

Friday, August 13, 2010

Collecting User Actions with GWT

While I was at one of the Google I/O GWT sessions (courtesy of Bizo), a Google presenter mentioned how one of their internal GWT applications tracks user actions.


The idea is really just a souped-up, AJAX version of server-side access logs: capturing, buffering, and sending fine-grained user actions up to the server for later analysis.


The Google team was using this data to make A/B-testing-style decisions about features–which ones were being used, not being used, tripping users up, etc.


I thought the idea was pretty nifty, so I flushed out an initial implementation in BizAds for Bizo’s recent hack day. And now I am documenting my wild success for Bizo’s first post-hack-day “beer & blogs” day.


No Access Logs


Traditional, page-based web sites typically use access logs for site analytics. For example, the user was on a.html, then b.html. Services like Google Analytics can then slice and dice your logs to tell you interesting things.


However, desktop-style one-page webapps don’t generate these access logs–the user is always on the first page–so they must rely on something else.


This is pretty normal for AJAX apps, and Google Analytics already supports it via its asynchronous API.


We had already been doing this from GWT with code like:


public native void trackInGA(final String pageName) /*-{
$wnd._gaq.push(['_trackPageview', pageName]);
}-*/;

And since we’re using a MVP/places-style architecture (see gwt-mpv), we just call this on each place change. Done.


Google Analytics is back in action, not a big deal.


Beyond Access Logs


What was novel, to me, about this internal Google application’s approach was how the tracked user actions were much more fine-grained than just “page” level.


For example, which buttons the user hovers over. Which ones they click (even if it doesn’t lead to a page load). What client-side validation messages are tripping them up. Any number of small “intra-page” things that are nonetheless useful to know.


Obviously there are a few challenges, mostly around not wanting to detract from the user experience:



  • How much data is too much?

    Tracking the mouse over of every element would be excessive. But the mouse over of key elements? Should be okay.



  • How often to send the data?

    If you wait too long while buffering user actions before uploading them to the server, the user may leave the page and you’ll lose them. (Unless you use a page unload hook, and the browser hasn’t crashed.)


    If you send data too often, the user might get annoyed.




The key to doing this right is having metrics in place to know whether you’re prohibitively affecting the user experience.


The internal Google team had these metrics for their application, and that allowed them to start out batch uploading actions every 30 seconds, then every 20 seconds, and finally every 3 seconds. Each time they could tell the users’ experience was not adversely affected.


Unfortunately, I don’t know what exactly this metric was (I should have asked), but I imagine it’s fairly application-specific–think of GMail and average emails read/minute or something like that.


Implementation


I was able to implement this concept rather easily, mostly by reusing existing infrastructure our GWT application already had.


When interesting actions occur, I have the presenters fire a generic UserActionEvent, which is generated using gwt-mpv-apt from this spec:


@GenEvent
public class UserActionEventSpec {
@Param(1)
String name;
@Param(2)
String value;
@Param(3)
boolean flushNow;
}

Initiating the tracking an action is now just as simple as firing an event:


UserActionEvent.fire(
eventBus,
"someAction",
"someValue",
false);

I have a separate, decoupled UserActionUploader, which is listening for these events and buffers them into a client-side list of UserAction DTOs:


private class OnUserAction implements UserActionHandler {
public void onUserAction(final UserActionEvent event) {
UserAction action = new UserAction();
action.user = defaultString(getEmailAddress(), "unknown");
action.name = event.getName();
action.value = event.getValue();
actions.add(action);
if (event.getFlushNow()) {
flush();
}
}
}

UserActionUploader sets a timer that every 3 seconds calls flush:


private void flush() {
if (actions.size() == 0) {
return;
}
ArrayList<UserAction> copy =
new ArrayList<UserAction>(actions);
actions.clear();
async.execute(
new SaveUserActionAction(copy),
new OnSaveUserActionResult());
}

The flush method uses gwt-dispatch-style action/result classes, also generated by gwt-mpv-apt, to the server via GWT-RPC:


@GenDispatch
public class SaveUserActionSpec {
@In(1)
ArrayList<UserAction> actions;
}

This results in SaveUserActionAction (okay, bad name) and SaveUserActionResult DTOs getting generated, with nice constructors, getters, setters, etc.


On the server-side, I was able to reuse an excellent DatalogManager class from one of my Bizo colleagues (unfortunately not open source (yet?)) that buffers the actions data on the server’s hard disk and then periodically uploads the files to Amazon’s S3.


Once the data is in S3, it’s pretty routine to setup a Hive job to read it, do any fancy reporting (grouping/etc.), and drop it into a CSV file. For now I’m just listing raw actions:


-- Pick up the DatalogManager files in S3
drop table dlm_actions;
create external table dlm_actions (
d map<string, string>
)
partitioned by (dt string comment 'yyyyddmmhh')
row format delimited
fields terminated by '\n' collection items terminated by '\001' map keys terminated by '\002'
location 's3://<actions-dlm-bucket>/<folder>/'
;

alter table dlm_actions recover partitions;

-- Make a csv destination also in S3
create external table csv_actions (
user string,
action string,
value string
)
row format delimited fields terminated by ','
location 's3://<actions-report-bucket/${START}-${END}/parts'
;

-- Move the data over (nothing intelligent yet)
insert overwrite table csv_actions
select dlm.d["USER"], dlm.d["ACTION"], dlm.d["VALUE"]
from dlm_actions dlm
where
dlm.dt >= '${START}00' and dlm.dt < '${END}00'
;

Then we use Hudson as a cron-with-a-GUI to run this Hive script as an Amazon Elastic Map Reduce job once per day.


Testing


Thanks to the awesomeness of gwt-mpv, the usual GWT widgets, GWT-RPC, etc., can be doubled-out and testing with pure-Java unit tests.


For example, a method from UserActionUploaderTest:


UserActionUploader uploader = new UserActionUploader(registry);
StubTimer timer = (StubTimer) uploader.getTimer();

@Test
public void uploadIsBuffered() {
eventBus.fireEvent(new UserActionEvent("someaction", "value1", false));
eventBus.fireEvent(new UserActionEvent("someaction", "value2", false));
assertThat(async.getOutstanding().size(), is(0)); // buffered

timer.run();
final SaveUserActionAction a1 = async.getAction(SaveUserActionAction.class);
assertThat(a1.getActions().size(), is(2));
assertAction(a1, 0, "anonymous", "someaction", "value1");
assertAction(a1, 1, "anonymous", "someaction", "value2");
}

The usual GWT timers are stubbed out by a StubTimer, which we can manually tick via timer.run() to deterministically test timer-delayed business logic.


That’s It


I can’t say we have made any feature-altering decisions for BizAds based on the data gathered from this approach yet–technically its not live yet. But it’s so amazing that surely we will. Ask me about it sometime in the future.

hackday: analog meters

For this last hackday, I decided to work on something more hardware hacking related. At this year's Maker Fair, I was really inspired by all the cool stuff people were building, so I picked up an arduino and started playing around with a couple of things.

I've always wanted to have some cool old-school analog VU type meters displaying web requests.

Here's my completed hackday project:



Here's a view of the components from the back:



It's battery operated and receives data wirelessly over RF from another arduino I have hooked up via serial to my laptop.

It's pretty simple, but I'm still totally psyched about how it came out.

The main components are some analog panel meters (kinda pricey, but awesome), and an RF receiver. The frame is a piece of scrap acrylic from TAP Plastics that I drilled and cut to size, and the stand is a piece of a wire clothes hanger bent to shape.

Connected to my computer is a another arduino (actually a volksduino) that receives updates over USB and sends the data out over RF:



You may be asking, why bother with wireless if you need a computer hooked up through serial anyway. Or you may ask why not just connect to a wireless network directly.

Well, I wanted the meters to be able to be moved around, or mounted on a wall... I wanted them wireless. But, it turns out that wireless and even ethernet solutions for connecting an arduino to the internet directly are comparatively pretty expensive. Even using bluetooth is expensive. My long term plan is to have a single arduino connected to the internet directly (via ethernet or wireless), and have it serve as a proxy over RF for the others... So this is a bit of work towards that.

I wrote a bit of Java code to connect to amazon's cloudwatch to pull the load balancer statistics for two of our services. I then discovered it's near impossible to connect to anything over USB in Java... It is ridiculous. Luckily, it's REALLY easy to do this with Processing, so I wrote a simple processing program that used my cloudwatch library and wrote it out to serial.

And that's really it. The arduino reads data over serial, and periodically sends it over RF. The arduino hooked up to the meters simply reads the values over RF and sets the meters to display a scaled version of the results. They're showing requests per second. We get a huge amount of requests per second with these services, so the numbers on the dial aren't actually correct (I need to make some custom faceplates). It also flashes an LED every time it gets a RF transmission.

Here's a quick video of it in action:




The one thing I'm not crazy about is that the maximum resolution you can get from cloudwatch is stats per minute, so the meters don't actually change as often as I would like.

Still, pretty cool. I'm looking forward to building some more displays like this in the future.

Thursday, July 29, 2010

Extending Hive with Custom UDTFs

Let’s take a look at the canonical word count example in Hive: given a table of documents, create a table containing each word and the number of times it appears across all documents.


Here’s one implementation from the Facebook engineers:



CREATE TABLE docs(contents STRING);

FROM (
MAP docs.contents
USING 'tokenizer_script'
AS
word,
cnt
FROM docs
CLUSTER BY word
) map_output
REDUCE map_output.word, map_output.cnt
USING 'count_script'
AS
word,
cnt
;

In this example, the heavy lifting is being done by calling out to two scripts, ‘tokenizer_script’ and ‘count_script’, that provide custom mapper logic and reducer logic.


Hive 0.5 adds User Defined Table-Generating Functions (UDTF), which offers another option for inserting custom mapper logic. (Reducer logic can be plugged in via a User Defined Aggregation Function, the subject of a future post.) From a user perspective, UDTFs are similar to User Defined Functions except they can produce an arbitrary number of output rows for each input row. For example, the built-in UDTF “explode(array A)” converts a single row of input containing an array into multiple rows of output, each containing one of the elements of A.


So, let’s implement a UDTF that does the same thing as the ‘tokenizer_script’ in the word count example. Basically, we want to convert a document string into multiple rows with the format (word STRING, cnt INT), where the count will always be one.


The Tokenizer UDTF


To start, we extend the org.apache.hadoop.hive.ql.udf.generic.GenericUDTF class. (There is no plain UDTF class.) We need to implement three methods: initialize, process, and close. To emit output, we call the forward method.


Adding a name and description:



@description(name = "tokenize", value = "_FUNC_(doc) - emits (token, 1) for each token in the input document")
public class TokenizerUDTF extends GenericUDTF {

You can add a UDTF name and description using a @description annotation. These will be available on the Hive console via the show functions and describe function tokenize commands.


The initialize method:



public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException

This method will be called exactly once per instance. In addition to performing any custom initialization logic you may need, it is responsible for verifying the input types and specifying the output types.


Hive uses a system of ObjectInspectors to both describe types and to convert Objects into more specific types. For our tokenizer, we want a single String as input, so we’ll check that the input ObjectInspector[] array contains a single PrimitiveObjectInspector of the STRING category. If anything is wrong, we throw a UDFArgumentException with a suitable error message.



if (args.length != 1) {
throw new UDFArgumentException("tokenize() takes exactly one argument");
}

if (args[0].getCategory() != ObjectInspector.Category.PRIMITIVE
&& ((PrimitiveObjectInspector) args[0]).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING) {
throw new UDFArgumentException("tokenize() takes a string as a parameter");
}

We can actually use this object inspector to convert inputs into Strings in our process method. This is less important for primitive types, but it can be handy for more complex objects. So, assuming stringOI is an instance variable,



stringOI = (PrimitiveObjectInspector) args[0];

Similarly, we want our process method to return an Object[] array containing a String and an Integer, so we’ll return a StandardStructObjectInspector containing a JavaStringObjectInspector and a JavaIntObjectInspector. We’ll also supply names for these output columns, but they’re not really relevant at runtime since the user will supply his or her own aliases.



List fieldNames = new ArrayList(2);
List fieldOIs = new ArrayList(2);
fieldNames.add("word");
fieldNames.add("cnt");
fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
fieldOIs.add(PrimitiveObjectInspectorFactory.javaIntObjectInspector);
return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
}

The process method:



public void process(Object[] record) throws HiveException

This method is where the heavy lifting occurs. This gets called for each row of the input. The first task is to convert the input into a single String containing the document to process:



String document = (String) stringOI.getPrimitiveJavaObject(record[0]);

We can now implement our custom logic:



if (document == null) {
return;
}
String[] tokens = document.split(“\\s+”);
for (String token : tokens) {
forward(new Object[] { token, Integer.valueOf(1) });
}
}

The close method:



public void close() throws HiveException { }

This method allows us to do any post-processing cleanup. Note that the output stream has already been closed at this point, so this method cannot emit more rows by calling forward. In our case, there’s nothing to do here.


Packaging and use:


We deploy our TokenizeUDTF exactly like a UDF. We deploy the jar file to our Hive machine and enter the following in the console:



> add jar TokenizeUDTF.jar ;
> create temporary function tokenize as ’com.bizo.hive.udtf.TokenizeUDTF’ ;
> select tokenize(contents) as (word, cnt) from docs ;

This gives us the intermediate mapped data, ready to be reduced by a custom UDAF.


The code for this example is available in this gist.

Friday, June 25, 2010

Come work at Bizo

Want to work on some interesting problems with a great development team?

We're looking to hire a junior developer and a quantitative engineer.

Send us your resume! Or if you know someone in the Bay Area that you think might be a good fit, please forward the posting to them.