Gluster and hadoop: Using FUSE options to defy the CAP theorem.

The bigtop smoke tests are a great way to hit your cluster with a broad range of hadoop workloads. 

One of the things I learned from running the diverse workloads that they offer for hadoop on top of gluster, is that gluster is really modular.  Something really cool about it is that you can change its consistency model and thus its performance, really easily - without even touching the translator stack.

One of the recent hidden gems which I've uncovered is the fact that the LDA mahout test creates 1000s of small files, thus comprising you to do a small file workload performance benchmark. 

Well, we found that in glusterfs-hadoop,

• if you vie for strict consistency by setting timeout attrs to 0 at mount time... you get burnt on small file performance. 
• BUT if you go for eventual consistency : you run into task failures with renames that occur on mapreduce workloads.

So... what do we do now? Well... theres another option: You can have both ! Just create TWO gluster mounts :) :) :) ...  one for mapreduce, and one for smallfile workloads.  In the end, the data all goes to the same FS, but you get much faster import of data if you dont force strict consistency. 

Okay so this is great.  But how do we tell hadoop which one to use?

I hate maintaining XML files.  I certainly dont want to maintain 2 XML files !  ... Hadoop runtime options  to the rescue. 

One of the lesser used hadoop options is runtime parameters.  For example, in glusterfs hadoop, we specify some XML parameters in our config files:

mount -o entry-timeout=0,attribute-timeout=0/mnt/glusterfsCONSISTENT

<property>
<name>fs.glusterfs.mount</name>
<value>/mnt/glusterfsCONSISTENT</value>
</property>

But alas, we can send these same parameters to the hadoop runtime file system command, so that we can have it write to a different mount:

    
    mount -o entry-timeout=2,attribute-timeout=2 /mnt/glusterfsFAST

hadoop fs -Dfs.glusterfs.mount=/mnt/glusterfsFAST -put /tmp/mylocaldir/* /dfs

So whats happening above ?  Well, we have one file system mount which is strictly consistent, and is used in the default hadoop configuration (which is launched in typical mapreduce jobs).  But for a specific case where we dont mind loose consistency -- to put several 1000 files, we create a FAST mount, with timeout options a little looser.  Initially, this seems to result in 8-10 X faster puts of files into gluster. 

Its nice that gluster plays so nicely with fuse, it allows you to do alot of interesting things both in the hadoop interop, and in general, for supporting multiple real world workflows at once in a single distributed file system.

Its also nice that hadoop lets you send it arbitrary configuration changes at runtime.  This means that for differing hadoop workloads, we can use differing FUSE mounts to hack around different performance bottlenecks.

This is just the beggining of my experimentation with the mount options, but to see others, check here:  http://gluster.org/community/documentation/index.php/Gluster_3.2:_Setting_Volume_Options (if theres an updated 3.3 doc, somebody let me know ill correct his link) .

Some GraphViz Templates for my OpenStack Savanna talk

Okay...  So I keep finding myself googling for these GraphViz templates.  Here's the ones im using for my OpenStack Savanna preso this tuesday.

This should be interesting to anyone who is either (1) interested in giving an openstack savanna talk or (2) stealing my slides and modifying them or (3) needs to spice up their graphviz chops.

UML style diagrams

node_group_template_master -> cluster_template [arrowhead = diamond];
node_group_template_slave -> cluster_template [arrowhead = diamond];
 cluster_template -> cluster

Subgraphs

Here's how to make subgraphs in.  I've used this to show the three stages of oozie. In this case, a super distilled subgraph : Russian dolls of the oozie layers.  

digraph G {
      subgraph cluster_a {
        Bundle;
    subgraph cluster_0 {   
           coordinator;
           subgraph cluster_1 {
           workflow;}}}}

ER type cardinality diagrams

And finally, you can use different "arrowhead" implementations "crow",  "diamond", and "empty".

 digraph cont  {
   cluster_group -> node_group [arrowhead = crow];
}
 

Yarn is an APP : Recognize !

This post is very raw... I'll refine it later....... But for those that run into ClassNotFound/MethodNotFound issues, it might be useful food for thought. 

Poking around in AMBARI recently, I found that there is this parameter that defines the yarn default classpath.  

yarn.application.classpath

SO WHAT you say... OF COURSE YARN HAS A CLASSPATH.  Who cares???

Well ... I care !

The reason why is that I always seem to run into MethodNotFound issues.  Especially when hadooping.  I wrote up my last MethodNotFound adventure in hadoop here.  Why ? Because cassandra seemed to be bundling its own obsolete version of avro in my classpath.  So I'm very dubious of the global hadoop lib/ classpath . 

Even though Im happy that yarn and mapreduce are decoupled:  But afraid that yarn has so much brought into its classpath by default.......... See below.

...( Note my fears here might be unjustified ... but lots of jars on the classpath usually spells nightmares for me)...

yarn.application.classpath	$HADOOP_CONF_DIR,$HADOOP_COMMON_HOME/share/hadoop/common/*,$HADOOP_COMMON_HOME/share/hadoop/common/lib/*,$HADOOP_HDFS_HOME/share/hadoop/hdfs/*,$HADOOP_HDFS_HOME/share/hadoop/hdfs/lib/*,$HADOOP_YARN_HOME/share/hadoop/yarn/*,$HADOOP_YARN_HOME/share/hadoop/yarn/lib/*	CLASSPATH for YARN applications. A comma-separated list of CLASSPATH entries

Anyways ... The old "/usr/lib/hadoop/lib" directory is now not so univeral as it used to be. 

What this means for people migrating to YARN ? 

You need to be smart about where you put your libs.  Hadoop is "sort of" OSGi (i.e. YARN now has its own libraries...) and thats a good thing - it means less likeliehood of hadoop apps tripping over differences in the hadoop library jars which they inherit classpath from.

But the downside? 

Now mapreduce apps need to make sure that, if you were preloading your hadoop/lib directory with some libraries for runtime deps, that you either (1) add that lib/ tothe yarn.application.classpath OR ELSE that you change where you are putting those jars (i.e. put them in /usr/lib/hadoop-yarn/lib  .  Im not sure what the exact right deployment idiom is, just yet... So dont qoute me.  

Just recognize that YARN classpath might not, by default, inherit from /usr/lib/hadoop/lib.

Anyways...

Will update this post once I learn more.  FYI For those of you using AMBARI, I think the ambari  folks have been gracious enough to include /usr/lib/hadoop/lib into the default yarn classpath.  This is good (if you are lazy), but might be bad (if you hate runtime NoSuchMethodERrors that tend to occur when you have the same version of a jar floating around in your class path + cutting edge API usage in your Apps).

Moral of the story: 

At the very least : You should know that yarn apps have a default classpath, and that hadoop/lib/  isn't necessarily on it.

Adding new virtual disks to running virt images.

I always finding my self clicking UI's and modifying command line incantations to create VMs with new virtual block devices.  This is probably pretty common to anyone who is using VMs to mimic a real world environment where there is a dedicated device for used by a particular program (in my case, its usually for creating a VM that tests a  gluster setup and needs a dedicated brick).

So here's how to add a new disk image into a running kvm instance using virsh (the shell for libvirt) and qemu-img.

First, create your disk on the host:

qemu-img create -f raw /VirtualMachines/idh2.img 10G

Now, we want to attach it.  Attach takes a "--target" option - and its important to name the target deliberately.  http://builder.virt-tools.org/artifacts/libvirt-virshcmdref/html/sect-attach-disk.html indicates that 

vdX -> becomes a virtio attachment

sdX -> becomes a scsi attchement

hdX -> becomes an IDE  attachment

So lets do virtio : I'm assuming its the easy way to attach devices for this scenario (i could be wrong) , but the sdX (scsi) approach for me results in a "Unable to determine model for scsi controller" error... So lets just go with vdx as the device name.  That will map to virtio: 

sudo virsh attach-disk rhs_21_gluster2 /VirtualMachines/idh2.img --target vda --driver qemu

        ...(you might need the --subdriver qcow2 option, not sure)...
Okay ! Now, lets make sure we *(really)* have a 10 Gig disk in there: 

[root@idh1 ~]# fdisk -l | grep vd*
Disk /dev/vdc: 10.7 GB, 10737418240 bytes 

Yay it worked !  So its really easy using qemu and virsh attach-disk to add new disk images at runtime.  Now Im not sure why it got called "vdc" instead of "vdX". But thats okay.  I now have my virtual block happily residing in my newly created VM... And I can easily automate stuff on top of this, and add more over time if I needed it without relying on the Virtual Machine Manager UI. 

Using jar -tf + mvn dependency:tree to escape jar-hell

mvn dependency:tree + jar -tf are your friends when tracking down redundant classes with subtle differences that break your code at runtime.

NOTE THAT IF YOU RUN INTO ISSUES LIKE THIS , IT IS LIKELY BECAUSE YOU HAVENT ENCAPSULATED YOUR RUNTIME PROPERLY.  In a hadoop cluster, you shouldnt expect crunch, pig, hive, ... to all run in the same classpath.  For example:  I ran into this in an eclipse project where I was attempting to add all the ecosystem tool APIs into the same module.  In retrospect, even though it was fixable, I really shouldn't have expected this to work in the first place.  Normal hadoop deployments, after all, have different lib/ directories for different tools - sort of in the OSGi sense, and that is the way also that eclipse packages its plugins - thus - java dependencies are better of being namespaced and isolated, rather than painstakingly resolved and integrated.

Nevertheless, its worth the effort to try and debug something like this, at least once, just so you know how to do it ...

... So here goes...
 
So, I recently hit a a cryptic "init()<V> not found" exception when running crunch in a maven project with Avro serialization.   This is by no means "avro" or "hadoop" or "crunch" specific:  Libraries like avro, which are used in alot of places, can be very dangerous shifting sands in your runtime environment.  Other culprits for jar hell are things like utilities (commons, guava, guice, logging frameworks, etc...). But regardless of where your jar-hell comes from, it always looks a little somethin like this:

java.lang.NoSuchMethodError: org.apache.avro.mapred.AvroKey: method <init>()V not found
    at org.apache.crunch.types.avro.AvroKeyConverter.getWrapper(AvroKeyConverter.java:57)
    at org.apache.crunch.types.avro.AvroKeyConverter.outputKey(AvroKeyConverter.java:36)
    at org.apache.crunch.types.avro.AvroKeyConverter.outputKey(AvroKeyConverter.java:25)
    at org.apache.crunch.impl.mr.emit.MultipleOutputEmitter.emit(MultipleOutputEmitter.java:41)
    at org.apache.crunch.MapFn.process(MapFn.java:34)
    at org.apache.crunch.impl.mr.run.RTNode.process(RTNode.java:99)
    at org.apache.crunch.impl.mr.emit.IntermediateEmitter.emit(IntermediateEmitter.java:56)
    at org.apache.crunch.MapFn.process(MapFn.java:34)
    at org.apache.crunch.impl.mr.run.RTNode.process(RTNode.java:99)
    at org.apache.crunch.impl.mr.emit.IntermediateEmitter.emit(IntermediateEmitter.java:56)
    at org.apache.crunch.MapFn.process(MapFn.java:34)
    at org.apache.crunch.impl.mr.run.RTNode.process(RTNode.java:99)
    at org.apache.crunch.impl.mr.run.RTNode.process(RTNode.java:110)
    at org.apache.crunch.impl.mr.run.CrunchMapper.map(CrunchMapper.java:60)
    at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:370)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:212)

In my case, older avro versions didn't have an empty argument contructor which crunch depends on.  It was very frightening at first site.... But after scanning the apache/crunch github repo, and checking both 1.3 and 1.5 avro branches, it became obvious that a change had occured to newer avro libraries which added a default empty constructor to the AvroKey class.

Now, I'll do my best for the rest of this post not to dive into the details of my specific problem - as jar hell is a very generic problem and (i think) the below strategy is a pretty generic strategy for solving any kind of jar hell problem.

How to fix it: high level.

The tricky part of all this is that there really is no way of easily knowing, at runtime, what jar is being used to dial up a particular class: the stack trace will always be the same.  Thus, you have to

- scan all JARS on your classpath for the redundant class using "jar -tf | grep <CLASS_NAME>.class".

- decide which jar you DONT want (usually this can be done by mapping the jar version to the code).

- find the offending parent pom dependency is pulling in the unwanted jar using "mvn dependency:tree"

- exclude it using <exclude> tags.

Detailed example:

scan all JARS

First, cd to where your jars live:

1)  cd to M2_HOME (this is normally $HOME/.m2 if you dont set it explicitly.

2) for f in `find . -name '*avro*.jar'` ; do echo "checking $f" && jar -tf $f | grep "AvroKey.class" ; done

          decide which jar you DONT want 

It should be obvious now which "culprit" jar you are looking at, for example, in my case:

jays-MacBook-Pro:.m2 jayunit100$ for f in `find . -name '*avro*.jar'` ; do echo "checking $f" && jar -tf $f | grep "AvroKey.class" ; done
checking ./repository/com/twitter/parquet-avro/1.2.0/parquet-avro-1.2.0-sources.jar
checking ./repository/com/twitter/parquet-avro/1.2.0/parquet-avro-1.2.0.jar
checking ./repository/org/apache/avro/avro/1.7.4/avro-1.7.4.jar
checking ./repository/org/apache/avro/avro-mapred/1.7.4/avro-mapred-1.7.4.jar
org/apache/avro/mapred/AvroKey.class
checking ./repository/org/apache/avro/trevni-avro/1.7.4/trevni-avro-1.7.4.jar
checking ./repository/org/apache/cassandra/deps/avro/1.4.0-cassandra-1/avro-1.4.0-cassandra-1-sources.jar
checking ./repository/org/apache/cassandra/deps/avro/1.4.0-cassandra-1/avro-1.4.0-cassandra-1.jar
org/apache/avro/mapred/AvroKey.class

Obviously, in my case what was happening is that somehow cassandra was packaging a version of avro up which had an old version of AvroKey.class,...

 3)  find the offending parent pom dependency

The next step is to find out WHERE maven is pulling that jar in.  To do that, you can use mvn dependency:tree.  This will show you exactly what top level pom dependency is responsible for bundling the naughty jar file.

[INFO] +- org.apache.mahout:mahout-examples:jar:0.8:compile
[INFO] |  +- org.apache.mahout:mahout-integration:jar:0.8:compile
[INFO] |  |  +- org.apache.solr:solr-commons-csv:jar:3.5.0:compile
[INFO] |  |  +- org.mongodb:mongo-java-driver:jar:2.11.1:compile
[INFO] |  |  +- org.mongodb:bson:jar:2.11.1:compile
[INFO] |  |  +- org.apache.cassandra:cassandra-all:jar:1.2.5:compile
[INFO] |  |  |  +- net.jpountz.lz4:lz4:jar:1.1.0:compile
[INFO] |  |  |  +- com.ning:compress-lzf:jar:0.8.4:compile
[INFO] |  |  |  +- com.googlecode.concurrentlinkedhashmap:concurrentlinkedhashmap-lru:jar:1.3:compile
[INFO] |  |  |  +- org.apache.cassandra.deps:avro:jar:1.4.0-cassandra-1:compile

Aha !  So its mahout's fault :) Now we can exclude it  

The final step now is to enable an exclusion filter which removes the above jar from the transitive dependencies which are pulled in via maven.  In my case, I chose to exclude "cassandra-all" since I don't want mahout to willy-nilly bundle anything cassandra related (i'd rather just have that in my top level pom to begin with).

<dependency>
            <groupId>org.apache.mahout</groupId>
            <artifactId>mahout-core</artifactId>
            <version>0.8</version>
            <exclusions>
                <!--
                    cassandra bundles an old avro, which conflicts with crunch's need
                    for the new avro, see
                    http://stackoverflow.com/questions/20951839/how-to-trace-the-origin-of-initv-failures-in-avro
                 -->               
                <exclusion>
                    <artifactId>cassandra-all</artifactId>
                    <groupId>org.apache.cassandra</groupId>
                </exclusion>
            </exclusions>
        </dependency>

Fixed ! Now, when you build your project, you can expect that older jars referenced wont be transitively included into your classpath.  

Of course, this all assumes that you dont NEED references to the older jar.  In that case, the jar-hell problem is much more difficult to solve: probably you will need to run your apps in different JVMs, for example. 

Ajax + Maven + S3 = A lightweight maven repository browser that never goes down.

Heavyweight interfaces sometimes occlude the simple, platform neutral, static directory structure that made maven famous to begin with.  If your server goes down, your jars are unavailable.  S3 solves this problem by providing serverless, http accessible storage.  Adding Javascript + ajax into the mix means you can have dynamic content hosted through s3 with zero administration. 

Problem: You have a public maven repository somewhere (i.e. in the cloud) - but you want to be able to browse it in a user freindly way.

Solution 1: Use a full blown repo manager archiva/artifactory/nexus to host it.  

There are use cases for this... i.e., you are a huge company with 100s of developers and you need a scalable interface to host lots of jar artifacts with many versions in a secure manner, and you also need to provide different users different permissions to maintain different repositories, etc...  But for a simple maven repo, some might say that a heavy weight repo manager creates a new problem:  You are maintaining a server just to host a couple of jar files.  The snazzy UI to your jar files is nice - but is it really necessary?  Do you really need "roles" and hosted md5 comparisons just to share a few binaries?  Probably not.

Now, Admittedly, for big organizations with security issues, etc, a private archiva server is a nice solution.... but probably isnt required for smaller projects which need a simple landing page for their maven artifacts for end users to consume.

Solution 2:  Put the maven repo in S3 and use JQueries snazzy XML parser to host a dynamic, HTML5+jquery/javascript front-end.  Here's how: 

1) Either use an s3 publisher in your "mvn deploy" invocations, or else, just scp your maven repository into s3://yourbucket/maven/

2) Now, you can look into http://<your_bucket>.s3.amazonaws.com/maven/repositories/internal/<your_repo_path>/maven-metadata.xml", and you will see that there is a nice layout of all versions of your Jar. For example: http://repo.maven.apache.org/maven2/org/apache/pig/pigsmoke/maven-metadata.xml contains several version tags. 

3) Save the code below to s3://<your_bucket>/maven/index.html

<html>
<head>
<title>GlusterFS-Hadoop Releases</title>

<script type="text/javascript"
 src="http://code.jquery.com/jquery-1.10.2.min.js"></script>

</head>
<body>
 <p> Releases </p>
 <div id="releases">
 </div> 
 <script type="text/javascript">
  
  $(document).ready(function(){
   $("#msgid").html("--");
  });
  
  function UrlExists(url, cb){
      jQuery.ajax({
          url:      url,
          dataType: 'text',
          type:     'GET',
          complete:  function(xhr){
              if(typeof cb === 'function')
                 cb.apply(this, [xhr.status]);
          }
      });
  }
 
  $(document).ready(function(){
    $("#releases").append("<ul></ul>");
    $.ajax({
      type: "GET",
      url: " http://rhbd.s3.amazonaws.com/<PATH_TO_YOUR_REPO>/maven-metadata.xml",
      dataType: "xml",
      success: function(xml){
      $(xml).find('metadata').each(function(){
        var versioning = $(this).find('versioning').text();
        var versions = $(this).find('versions').text();
        
        $.each(versions.split("\n"),
          function(index,value ) {
           var jarlink = "http://<YOUR_BUCKET_NAME>.s3.amazonaws.com/maven/<PATH_TO_YOUR_REPO>/VERSION/<YOUR_ARTIFACT_NAME>-VERSION.jar"
           var regex = new RegExp('VERSION', 'g');
           var jarversion = jarlink.replace(regex,value.trim());
           //$("<li></li>").html(jarversion).appendTo("#releases ul");
           $("<li></li>").html("<a href=\""+jarversion+"\">"+value+"</a>").appendTo("#releases ul");
        }
        );
      });
    },
    error: function() {
      alert("An error occurred while processing XML file ");
    }
    });
  });

 </script>
 
</body>
</html>

4) Now, browsing to http://<YOUR_BUCKET>.s3.amazonaws.com/index.html , You will see a nice list of links to your maven jars.  This is super easy to modify and maintain, and if using a object store like s3, you can gaurantee that your users will always have access to your artifacts.  Moral of the story: You can easily modify your landing page to your maven jars without using a heavy weight repo manager.


Checkout our glustersf-hadoop landing page http://rhbd.s3.amazonaws.com/maven/index.html, as an example of how this works... as long as you continue to use "mvn deploy:deploy" to update your repo, the HTML page will always serve the latest contents, because the maven-metadata.xml page will be automatically updated in the publishing process.

Building a Hive table over multideliimiter CSV Files.

One of the goals in the BigPetStore project is to demonstrate how to process the same data set using multiple different ecosystem components.

Processing of the BigPetStore input data set is described in this post, using pig.


The basic data format (subject to change over time) is intentionally a little complex:

BigPetStore,storeCode_AK,1    watt,duane,Tue Dec 30 02:42:47 EST 1969,48.25,flea collar
BigPetStore,storeCode_AK,2    stevens,sanford,Thu Jan 01 00:01:50 EST 1970,66.85,premium cat food

In pig, we load the data set as is, and then deal with merging the two comma separated components into one big pig tuple by "flattening" the first segment, and then "flattening" the second segment, yielding something like this:

BigPetStore    storeCode    AK    1    watt    duane    Tue Dec 30 02:42:47 EST 1969    48.25    flea collar

In hive, I decided to try a different approach: Using the RegExSerde class.

A quick note on libraries before you try this:

The RegExSerde class allows you to load heterogenously delimited data, its pretty simple to implement.  I had to make sure that the hive-serde and hive-contrib jars were in my HADOOP_HOME/lib/ path (in pure hive, you can just issue an "add" command to add jars to the classpath at runtime, but in bigpetstore, we do everything in the IDE and in local hadoop mode, so I manually added these jars into HADOOP_HOME/lib/ to get it working from my IDE). 

So : Here's how to define a Hive table over a file with more than one delimiter.

1) Craft a "base" definition for your Hive table the way you would any normal SQL table... Jot it down... Something like:

     String create =
                "CREATE TABLE hive_bigpetstore_etl (" +
                "  a1 STRING," +
                "  b2 STRING," +
                "  c3 STRING," +
                "  d4 STRING," +
                "  e5 STRING," +
                "  e6 STRING," +
                "  e7 STRING," +
                "  e8 STRING)" +
                "STORED AS TEXTFILE";

2) You build a regex where each capturing group corresponds to a column in your hive table.

3) Copy your regex into the input.regex field of your CREATE TABLE statement.

Hive will then consecutively put each group matched per row of data into successive columns defined in your CREATE TABLE statement.

Here is a snippet from how we use this in the BigPetStore project to create a table over the raw input data set without doing any manual ETL steps:

 String create =
                "CREATE TABLE hive_bigpetstore_etl (" +
                "  a1 STRING," +
                "  b2 STRING," +
                "  c3 STRING," +
                "  d4 STRING," +
                "  e5 STRING," +
                "  e6 STRING," +
                "  e7 STRING," +
                "  e8 STRING)" +
               
                " ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe' " +
                "WITH SERDEPROPERTIES  (" +
                "\"input.regex\" = \"INPUT_REGEX\" , " +
                "\"output.format.string\" = \"%1$s %2$s %3$s %4$s %5$s\") " +
                "STORED AS TEXTFILE";       

 Now, in my case, this regex was sufficiently complicated that I defined a "base" string that structured the "create" statement, and then layered the regex in programmatically.  You could do this in bash to, or else, even in hive with hive variables.  In my pure java hive create table statement, I did:

         create=create.replaceAll(
                 "INPUT_REGEX",
                 "(?:BigPetStore,storeCode_)" +
                 "([A-Z][A-Z])," + //state (CT)
                 "([0-9]*)" + // state transaction id (1)
                 "(?:\t)" + // [tab]
                 "([a-z]*)," + //fname (jay)
                 "([a-z]*)," + //lname (vyas)
                 "([A-Z][^,]*)," + //date starts with capital letter (MWTFSS)
                 "([^,]*)," + //price (12.19)
                 "([^,]*).*"); //product (premium cat food)
        System.out.println(create);
        ResultSet res =  stmt.executeQuery(create);

 And finally, we execute the query:

        res = stmt.executeQuery(
                "LOAD DATA INPATH '<rawInput>' INTO TABLE hive_bigpetstore_etl"
                    .replaceAll("<rawInput>", rawInput));

So there you have it:  Its easy to put a uniform, relational view over line delimited  data using Hive and the REGEX SerDe.... You just have to understand how capturing groups work !