Big Data,Spatial Pig,Threaded Visualization

This post is PACKED with goodies - One of the ways to analyze large sets of data in the Hadoop File System without writing MapReduce jobs is to use Apache Pig. I highly recommend that you read Programming Pig, in addition to the online documentation. Pig Latin, the scripting language of Pig, is easy to understand, write and more importantly to extend. Since we do spatial stuff, the first goodie extends Pig Latin with a spatial function when analyzing data from HDFS. Here is the problem I posed to myself, given a very large set of records containing an X and Y field, and given a set of polygons, I want to produce a set of tuples containing the polygon id and the number of X/Y records in that polygon. Nothing that a GP/Py task cannot do, but needed the exercise and BTW, you can call Pig from Python (post for another day).
Pig, when executing in MapReduce mode, converts the Pig Latin script that you give it into a MapReduce job jar that it submits to Hadoop. You can register additional jars that get packaged into the job jar and define UDFs (User Defined Function) to extend Pig Latin with new capabilities. The UDF that I wrote returns the polygon identifier that contains a given X and Y values. The polygon set is read by the UDF at startup from a shapefile that is loaded into the Distributed Cache. The polygons are spatially indexed for fast query and PIP (Point In Polygon) operation during tuple (x,y) evaluation. The Evaluator extends the org.apache.pig.EvalFunc class and is constructed with a reference to the shapefile path in the Hadoop file system.

    public PIPShapefile(final String path)
    {
        m_path = path; // Save reference as field variable.
    }

A mapping of the referenced shapefile to an internal name should be defined. The book Programming Pig dwells in detail on this.

    @Override
    public List getCacheFiles()
    {
        final List list = new ArrayList(1);
        list.add(m_path + ".shp#file.shp"); // Note the #
        return list;
    }

Since the evaluator return the polygon identifier, we tell the function to return an integer.

    @Override
    public Schema outputSchema(Schema input)
    {
        return new Schema(new Schema.FieldSchema(null, DataType.INTEGER));
    }

Finally, we implement the exec method that will be executed on every input tuple. This is a quick and dirty PIP implementation using JTS (leaving to the reader as an exercise to expand and elaborate :-)

    @Override
    public Integer exec(final Tuple tuple) throws IOException
    {
        m_init.init();
        int rc = -1;
        if (tuple != null && tuple.size() > 1)
        {
            double minArea = Double.POSITIVE_INFINITY;
            final Double x = (Double) tuple.get(0);
            final Double y = (Double) tuple.get(1);
            // log.info(x + " " + y);
            m_envelope.init(x - OFFSET, x + OFFSET, y - OFFSET, y + OFFSET);
            final List list = m_spatialIndex.query(m_envelope);
            for (final Feature feature : list)
            {
                if (feature.envelope.covers(x, y))
                {
                    final double area = feature.envelope.getArea();
                    if (area < minArea)
                    {
                        minArea = area;
                        rc = feature.number;
                    }
                }
            }
        }
        return rc;
    }

Note the m_init.init() invocation, this is because there is no initialization entry point in the lifecycle of a UDF. So the m_init is a reference to an implementation that mutates after the first time through from a shape file loader to an empty function. I used the geotools library to load and spatially index the features for later processing.

    private interface IInit
    {
        void init() throws IOException;
    }
    
    private final class FTTInit implements IInit
    {
        public void init() throws IOException
        {
            m_spatialIndex = new STRtree();
            final ShapefileReader shapefileReader = new ShapefileReader(new ShpFiles("file.shp"), true, true, m_geometryFactory);
            try
            {
                shapefileReader.disableShxUsage();
                while (shapefileReader.hasNext())
                {
                    final ShapefileReader.Record record = shapefileReader.nextRecord();
                    final Envelope envelope = new Envelope(record.envelope());
                    final Feature feature = new Feature();
                    feature.envelope = envelope;
                    feature.number = record.number;
                    m_spatialIndex.insert(envelope, feature);
                }
            }
            finally
            {
                shapefileReader.close();
            }
            m_init = new NoopInit(); // NOOP after first time through !!
        }
    }

    private final class NoopInit implements IInit
    {
        public void init()
        {
        }
    }

Compile the class using the following jars and package all into a jar to be registered with Pig.

gt-api-9-SNAPSHOT.jar (include)
gt-data-9-SNAPSHOT.jar (include)
gt-main-9-SNAPSHOT.jar (include)
gt-metadata-9-SNAPSHOT.jar (include)
gt-opengis-9-SNAPSHOT.jar (include)
gt-shapefile-9-SNAPSHOT.jar (include)
jts-1.13.jar (include)
pig-0.10.0.jar (exclude)

You can download the source code for PIPShapefile.java from here.

Before executing the script, I loaded the shapefile into HDFS using hadoop CLI commands:

$ hadoop fs -mkdir geo
$ hadoop fs -put cntry06.shp geo
$ haddop fs -put cntry06.dbf geo

I generated a million random points using AWK:

$ cat gendata.awk
BEGIN{
        OFS="\t"
        for(I=0;I < 1000000;I++){
                X=-180+360*rand()
                Y=-90+180*rand()
                print "L-"I,X,Y
        }
        exit
}

$ awk -f gendata.awk > data.tsv
$ hadoop fs -mkdir data
$ hadoop fs -put data.tsv data

The following is the Pig Latin script that solves the above stated problem; As you can see it is pretty self-explanatory:

$ cat script.pig

register '/Users/mraad_admin/JavaWorkspace/GeomX/out/GeomX.jar'
define pip com.esri.PIPShapefile('/user/mraad_admin/geo/cntry06');
A = LOAD 'data/data.tsv' AS (name:chararray,x:double,y:double);
B = FOREACH A GENERATE pip(x,y) AS P:int;
F = FILTER B BY P != -1;
G = GROUP F BY P;
C = FOREACH G GENERATE group,COUNT(F);
dump C;

Register the jar to be included in the MapReduce job jar. Define pip UDF with reference to the shape file in the HDFS. Load the data.tsv from HDFS with the described schema. Iterate over the data and generate a list of polygon ids based on x and y values. Keep all "found" polygon (id != -1), group them by the id value, and finally dump each id and its count. Pretty neat, no ?  To execute the script:

$ pig -f script.pig

Now, what to do with all these outputted tuples ? Goodie #2; Execute Pig on the server side and send the output to a client side app for visualization. Using Spring Data Hadoop, we can invoke a Pig script using a Pig Server instance and using Spring Flex expose it as a service for Remote Object invocation.
Here is the PigService implementation:

package com.esri;

@Service("pigService")
@RemotingDestination
public final class PigService
{
    @Autowired
    public PigServerFactoryBean m_pigServerFactoryBean;

    public List run(final String load, final String funcArg) throws Exception
    {
        final List list = new ArrayList();
        final PigServer pigServer = m_pigServerFactoryBean.getObject();
        try
        {
            pigServer.registerJar("/Users/mraad_admin/JavaWorkspace/GeomX/out/GeomX.jar");

            pigServer.registerFunction("pip", new FuncSpec("com.esri.PIPShapefile", funcArg));

            pigServer.registerQuery("A = load '" + load + "' as (name:chararray,x:double,y:double);");
            pigServer.registerQuery("B = foreach A generate pip(x,y) as P:int;");
            pigServer.registerQuery("F = FILTER B BY P != -1;");
            pigServer.registerQuery("G = GROUP F BY P;");
            pigServer.registerQuery("C = FOREACH G GENERATE group,COUNT(F);");

            final Iterator tupleIterator = pigServer.openIterator("C");
            while (tupleIterator.hasNext())
            {
                final Tuple tuple = tupleIterator.next();
                final ASObject asObject = new ASObject();
                asObject.put("index", tuple.get(0));
                asObject.put("count", tuple.get(1));
                list.add(asObject);
            }
        }
        finally
        {
            pigServer.shutdown();
        }
        return list;
    }
}

Notice how the Pig Script is broken down by registering the jar, the function, the queries and all are executed when openIterator is invoked, resulting in an iterator that enables me to convert the tuples into AS3Objects ready for AMF transfer to the client.

Final step; Invoke, fuse and visualize. The Flex API for ArcGIS is a great tool. In this post's implementation, I am taking advantage of one of the enhancements in the latest Flash Player (11.4); Workers. Finally, we have "threads" in the Flash Player! Take a look at Lee Brimelow video tutorial for a great description of worker usage. The application enables the user to load a local shapefile.  I am using a worker to parse the shapefile that was loaded into the distributed cache and convert each shape into a Graphic instance for visualization. The shapefile is a bit over 3.5MB, and in the "old" days that would have spun up the "beach ball of death" on my mac while parsing this size shapefile.  Now....no beach ball and the UI is still responsive. That was goodie #3. I had to play a bit of "games" by creating separate projects to enable the creation of the worker swf and the common code.  All this should be resolved in theory with the release of the Flash Builder 4.7. The Flex application enables the user to invoke the PigService and the resulting tuples are merged into the loaded feature attribute for thematic rendering. Goodie #4, ThematicFillSymbol is a custom symbol with full 180 warp-around polygon rendering.  It accepts as parameters the min/max count value and the min/max color value and fills each polygon proportionally to the count attribute value.

Well, that was a lot of stuff.  If you stayed with me all the way till here....thank you. Like usual, you can download all the source code of the server from here and the client from here.

MongoDB + Spring + Mobile Flex API for ArcGIS = Harmonie

I've used MongoDB on a project for the City of Chicago with great success.  I was impressed with the fact that we can store JSON documents in one giant collection, scale horizontally by just adding new nodes, the plethora of language APIs (Java,AS3) that can talk to it, run MapReduce tasks, and my favorite is that you can create a true spatial index on a document property.  This is not some BTree index on a compounded x/y properties, but a true spatial index that can be used to perform spatial operation such as near and within.  Today, Mongo only supports point geometries, but I understand that they are working on storing and spatially indexing lines and polygons and enabling other spatial operations. I've experimented with great success in an intranet in consuming BSON object directly from Mongo into a Flex based application, but the direct socket connection is a problem in a web or mobile environment. In addition, what I wanted was some middle tier that can turn my BSON documents into a full fledge Flex AGS Graphic instance with as little impedance as possible. When we wrote the Flex API for ArcGIS, we annotated key classes with "RemoteClass" to enable direct AMF streaming from ArcGIS Server at the SoC level.  What that means, is that there is no impedance mismatch between the object types created on the server with the object types consumed on the client. Some of these annotated classes are, FeatureSet, Graphic, MapPoint, SpatialReference. Whenever the words middle tier and AMF are together, you have to think BlazeDS with its RemoteObject capabilities on POJO and whenever you say POJO, the SpringFramework is by far the best at handling POJOs. Lucky for me, Under the Spring Framework umbrella exist two projects that the marriage of the two will create beautiful harmony; the Spring BlazeDS Integration and the Spring Data for MongoDB.  The BlazeDS Integration will handle the exposure of annotated POJOs as remote services that can be consumed from Flex and using IoC, these POJOs are injected with MongoDB templates for data access.  In today's world, client applications have to be of mobile nature and my experiment to fuze all these entities is no exception. So, I've written a simple Flex mobile application, that declares a RemoteObject with an http based endpoint to a server side registered POJO that exposes a "near" operation. On the client, the user can tap a location on the map.  The location is an argument to the "near" server side function, that invokes on MongoDB the "near" operation on a collection of tweet documents. The found tweet documents are converted to Graphic instances and returned back to the Flex application as part of a FeatureSet. The features in the FeatureSet are assigned to the graphicProvider property of a GraphicLayer, thus rendering these features on the map.  Just for coolness, I placed a switch toggle as a map control to enable/disable the clustering of the returned features. The key glue to this fusion is the mongo mapping converter implementation that converts BSON object to AGS features.


public class FeatureReadConverter implements Converter
{
    public Feature convert(final DBObject dbo)
    {
        final List shape = (List) dbo.removeField("shape");
        final Double lon = shape.get(0);
        final Double lat = shape.get(1);
        final MapPoint mapPoint = new MapPoint(
           WebMercator.longitudeToX(lon),
           WebMercator.latitudeToY(lat));
        final ASObject attributes = new ASObject();
        for (final String key : dbo.keySet())
        {
            attributes.put(key, dbo.get(key));
        }
        return new Feature(mapPoint, attributes);
    }
}

Here is a snapshot of the application in action from my iPhone !

And like usual, you can download the client source code from here, and the server side code from here.
Disclosure - The current flaring does not work on mobile using the released API unless you include the mx swc :-( this has been fix for the next release (don't ask....soon :-)

Big Data, Small Data, Big Visualization

Ever since I became a Cloudera Certified Developer for Apache Hadoop, I've been walking around with a hammer written on it "Map Reduce" looking for Big Data nails to pound.  Finally, a real world problem from a customer came to my attention where a Hadoop implementation will solve his dilemma. Given a 250GB (I know, I know, this is _not_ big) CSV data set of demographic data consisting of gender, race, age, income and of course location, and given a set of Point of Interest locations, generate a 50 mile heatmap for each demographic attribute for each the POI locations.
Using the "traditional" GeoProcessing with Python would take way more than a couple of days to run and would generate over 850GB of raster data. What do I mean by the "traditional" way ? You load the CSV data into a GeoDatabase and then you write an ArcPy script that; for each location, generate a 50 mile buffer.  Cookie cut the demographic data based on an attribute using the buffer and pass that feature set to the statistical package for density analysis which generates a raster file. Sure, you can manually partition the process onto acquired high CPU machines, but as I said, all that has to be done manually and will still take days.

There gotta to be a better way!

Enter Hadoop and a "different" way to process the data by taking advantage of:
- Hadoop File System fast splittable input streaming
- Distributed nature of Hadoop map reduce parts
- Distributed cache for "join" data
- External Fast java-based computational geometry library
- Producing vectors data rather than raster images

The last advantage is very important. This is something I call "Cooperative processing". See, people forget that there is a CPU/GPU on their client machines. If the server can producer vector data and we let the client render that data based on its capabilities, we will have a way more expressive application and the size of the data is way smaller. Will explain that in a bit.

Let me go back to the data processing. Actually, there is nothing to do.  There is no need for a transform and load process, as the CSV data can be directly placed onto an HDFS folder. The hadoop job will take as input the HDFS folder.

The Mapper Task - After instantiation, the 'configure' method is invoked to load the POI locations from the distributed cache and a 50 mile buffer is generated for each POI location using the fast computational geometry library, whereupon the buffer polygons are stored in an memory-based spatial index for fast intersection look up. The 'map' method is invoked on every record in the 250GB CSV input, where each record is tokenized for coordinates and demographic values. Using the coordinates and the prebuilt spatial index, we can find the associated POI locations.  Each 50 mile buffer is logically divided into kernel cells. Knowing the POI location, we can determine mathematically the relative kernel cell. We emit as a map key the combination of the POI location and the demographic value, and we emit the relative kernel cell as a map value.

map(k1,v1) - list(k2,v2)
where:
k1 = lineno
v1 = csv text
k2 = POI location,demographic
v2 = cellx,celly,1

Again, taking advantage of the powerful shuffle and sort capability of Hadoop on the POI Location/demographic key, I am ensuring that a reduce task will receive all the cells for a POI location/demographic combination.

The Reduce Task - For a POI location/demographic, the reduce method is invoke with the list of its associated cells. Cells with the same cellx,celly values are aggregated to produce a new list. We compose a JSON document of the new list and we emit the string representation of the JSON document using a custom output formatter onto which we override the 'generateFileNameForKeyValue' method to return something of the form "poi-location_demographic.json".

reduce(k2,list(v2)) - list(k3,v3)
where:
k2: POI location, demographic
v2: cellx, cellx, 1
k3: POI location, demographic
v3: JSON Text

I was able to validate my progress by invoking MRUnit on my codebase to ensure the soundness of my logic.

I packaged my map/reduce code and the geometry library into a jar, and I was ready to test it on the 250GB CSV.

But where to run this ?

Enter Amazon Elastic MapReduce. With a virtual swipe of a credit card, I was able to start up 10 large instances passing it a reference to my data and my jar in S3. 30 minutes later, a set of JSON files where produced in S3 occupying 238MB of space! Pretty cool, eh ? Compare that to days of execution time, and 850GB of rasters.  What is even more exciting, after a set of trials and errors and density kernel adjustments, I looked up my account balance and I owed Amazon $37.67 (will cost more to process a reimbursement request) !

Next comes the fun part, how to represent this JSON data for a particular POI location/demographic? Enter the Flex API for ArcGIS with its amazing extensibility and the flash player with its vector graphic and bitmap GPU-enhanced capabilities. See, by using the gradient filling of a drawn circle and the screen blending mode when placing that circle onto a bitmap, a set of close points will dissolve into a heatpoint. So, by taking advantage of this collaborative process between the server and the client, where the server generates a weighted point and let the client rasterize that point based on its weight, you get an expressive dynamic application. Let's push the visualization further to the coolest platform....the iPad. Flex code can be cross-compiled to run natively on the an iOS device. Let's push a bit more....3D. Taking advantage to the Stage3D capability, the heatmap vector data can be downloaded at runtime and dynamically morphed into a heightmap and a texture that can be draped on that heightmap. And here is the result....I call it "Heatmap in the Cloud". You can download the pdf of this Esri UC 2012 presentation from here. Have fun.

Enter the Fifth Dimension; Sentiment Map Rendering

We have been plotting x,y,z and time for a while now. But this is all syntax; Tim Berners-Lee believes that the future of the web is semantics, and I could not agree more. You have a glimpse of it today when you use for example Apple’s Siri. In our GIS space, MetaCarta at one time had a great appliance/service that inferred location from the semantics of a sentence. For example, “Joe ate a Chicago style pizza in downtown Boston”, would return the latitude and longitude of Downtown Boston and would ignore “Chicago” as a location, Impressive! With the explosion of social media and the infusion of geo-locations as ‘fields’ in our records, we are reduced to plotting locations with cute popups. Boring! We humans are quite the emotional animals (some more that others, specially if these humans come from the Middle East :-) and these emotions are quite visible now within our tweets and facebook posts. So, what if I can plot these sentiments, there could be something to “see” in these maps. Enter LinguaSys, I met one of their senior scientists by accident when sheltering from a rain storm. We ended up talking about the weather, this and that and of course the question “What do you do ?” had to eventually come up. “I do sentiment analysis”, he replied. “Wow, that is exactly what I was researching before leaving”, I replied. “I am looking for an ‘engine’ that I can pass it, say a set of tweets, and it will return to me for each tweet a sentiment index.”. “Not sure about tweets”, he replied, “as we deal with entire documents, but I am sure we can adjust our engine to such a process”. The rain stopped, we exchanged contact info and parted on the promise that we will stay in touch. A couple of month later, a very good customer with receptive avant-garde ideas needed something ‘new’. I proposed the sentiment index mapping based on social media to highlight areas of interest. The following is a derivative of this work based on a totally different interest; TSA approval or disapproval tweets. Working with LinguaSys, I was handed a set of XML files where one contained the tweets and associated sentiment indexes, for example:

<text>RT @msnbc_travel: Good news for elderly fliers (75 and above): TSA announces pilot program the relaxes security procedures http:\/\/t.co\/WTaM9AgW</text>
<disapprovalFactor factor="-0.8" reason=“indicatorOfSatisfaction”/>

Note that the factor is negative to indicate a level of satisfaction. The internal factor ranges from -1 (totally satisfied) to 1 (totally pissed off :-). Easy to parse and to associate a range based renderer. The second file is the “location” of the tweeters. Notice that I put location between quotes, that is because some values were great, like an exact latitude/longitude. Others were like “Boston, MA” And some (and most) were like “Earth”, or “Best Location, NYC!” or my favorite “Look behind u… Boo!”. There was a “sense” of location in these that I think would have given MetaCarta a run for its money. This is where being a unix CLI geek with tools like awk, grep and sed come to the rescue to massage the data. I downloaded a cities.csv to cross reference a city/state name to a location and now I can plot on a map the locatable tweets. Using the latest built-in capabilities of ArcGIS API for Flex such as clustering with flares, info window rendering on clicks, custom function referencing in symbols, I was able to quickly build an application to display the sentiments. You can see the application in action here. Hover over a cluster to flare it and click on a flare element to see its details. To make things more understandable, I reversed the factor value displayed in the info window. Warning: These are real tweets with sometimes very offensive language, so…. do not call HR on me, ok ? And like usual, you can see the source code here.
BTW, Check out GeoTagger for ArcGIS Runtime to see MetaCarta in action.

DnD File using HTML5 into Flex Web App

So….I always wanted to Drag and Drop an external file into a Flex application running in the browser. Well.. you can use the FileReference API to browse and load a file, but I really wanted to DnD. Unfortunately, this is not allowed due to security constraints of the Flash Player. However, using the latest proposed W3C DnD API and the File API, I should be able to use the last two and the AS3 ExternalInterface API to accomplish what I want. So, here is the content of a simple xml file that I would like to DnD and render on a map:

<markers>
    <marker x="45" y="45" label="M 1" info="This is M1"/>
    <marker x="-45" y="-45" label="M 2" info="This is M2"/>
</markers>

I modified the HTML wrapper to use the JavaScript DnD and File API (if available) to listening for “dragenter”, “dragover” and “drop” events on the FlashPlayer container.

var DnD = {
loadHandler:function () {
  var dropContainer = document.getElementById("DnDApp");
  dropContainer.addEventListener("dragenter", function (event) {
    event.stopPropagation();
    event.preventDefault();
  }, false);
  dropContainer.addEventListener("dragover", function (event) {
    event.stopPropagation();
    event.preventDefault();
    event.dataTransfer.dropEffect = 'copy';
  }, false);
  dropContainer.addEventListener("drop", function (event) {
    event.stopPropagation();
    event.preventDefault();
    var files = event.dataTransfer.files, len = files.length;
    for (var i = 0; i < len; i++) {
      var file = files[i];
      var fileReader = new FileReader();
      fileReader.onload = function (event) {
        dndApp.drop(event.target.result);
      }
      fileReader.readAsText(file);
    }
  }, false);
}
};
if( window.File && window.FileReader){
  window.addEventListener("load", DnD.loadHandler, false);
} else {
  alert('Your browser does not support File/FileReader !');
}

On dragenter, I stop the event propagation and prevent the default behavior. On dragover, I do the same and in addition, I update the drop effect to show a “+” icon over the drop area. And finally, on drop, I iterate over the list of dropped files, whereupon I read as text each file using the FileReader API. When the file is read (remember, this is all asynchronous), I hand over the content to the Flex application. On creation completion of the Flex application, the “drop” callback is registered using the EternalInterface enabling the host javascript wrapper to invoke the internal dropHandler function.

private function this_creationCompleteHandler(event:FlexEvent):void
{
  ExternalInterface.call("setObjectID",
    ExternalInterface.objectID);
  ExternalInterface.addCallback("drop", dropHandler);
}

private function dropHandler(text:String):void
{
  const doc:XML = new XML(text);
  for each (var markerXML:XML in doc.marker)
  {
    const mapPoint:MapPoint = new WebMercatorMapPoint(
      markerXML.@x,
      markerXML.@y);
    arrcol.addItem(new Graphic(
      mapPoint,
      null, {
        label: markerXML.@label,
        info: markerXML.@info }));
  }
}

The latter accepts a String argument that is converted into an XML instance, and using E4X, each child marker element is converted to a Graphic that is added to a graphic layer graphic provider array collection. Cool, eh ? Note that the graphic layer has it infoWindowRenderer property defined, is such a way that if you click on any of its graphics, a info window content will be displayed whose content is an instance of the defined component. Like usual all the source code is available here. Have fun DnD’ing.
You can see the application in action by download the markers.xml file, and drag and drop it on the application running here.
PS: As of this writing, the two JS APIs work in Google Chrome 16 and later, Firefox 3.6 and later, Safari 6 will support the standard File API and our favorite (Not!) Internet Explorer 10 (Preview 2+). One of these days, will come back to this and use something like Dojo DnD to abstract me from all this - that will be a nice post!