Since MapReduce (MR) is generally considered dead, or at least dying, it is very useful to know how to create Spark jobs from those existing in MR. The following method can be applied to any MR job. We will consider the GeoMesa Accumulo loading job described in the GeoMesa tutorial (geomesa-examples-gdelt) for this case.

An MR job is typically made up of three parts: the mapper, the reducer, and the driver. The GeoMesa example is a map-only job and therefore requires no reducer. The job takes a GDELT input line, creates a (Key,Value) pair from an empty Text object and the created GeoMesa SimpleFeature, and uses the GeoMesaOutputFormat to load the data to Accumulo. The full code of the MR job can be found in our repository; next this we will work through the key parts and suggest the changes required for Spark.

The job is initiated from the main method; the first few lines are related to parsing the required options from the command line, such as the Accumulo username and password. We then reach:

SimpleFeatureType featureType =
    buildGDELTFeatureType(featureName);
DataStore ds = DataStoreFinder.getDataStore(dsConf);
ds.createSchema(featureType);
runMapReduceJob(featureName, dsConf,
    new Path(cmd.getOptionValue(INGEST_FILE)));

The GeoMesa SimpleFeatureType is the primary mechanism used to store data in a GeoMesa data store and it needs to be initialized once, along with the data store initialization. Once this is done we execute the MR job itself. In Spark, we can pass the arguments via the command line as before, and then do the one-off setup:

spark-submit --class io.gzet.geomesa.ingest /
             --master yarn /
             geomesa-ingest.jar <accumulo-instance-id>
...

The contents of the jar contain a standard Spark job:

val conf = new SparkConf()
val sc = new SparkContext(conf.setAppName("Geomesa Ingest"))

Parse the command line arguments as before, as well as performing the initialization:

val featureType = buildGDELTFeatureType(featureName)
val ds = DataStoreFinder
   .getDataStore(dsConf)
   .createSchema(featureType)

Now we can load the data from HDFS, using wildcards if required. This creates one partition for each block of the file (64 MB default), resulting in an RDD[String]:

val distDataRDD = sc.textFile(/data/gdelt/*.CSV)

Or we can fix the number of partitions, depending upon our available resources:

val distDataRDD = sc.textFile(/data/gdelt/*.CSV, 20) 

Then we can perform the map, where we can embed the function to replace the process in the original MR map method. We create a tuple (Text,SimpleFeatureType) to replicate a (Key, Value) pair so that we can use the OutputFormat in the next step. When Scala Tuples are created in this way, the resulting RDD gains extra methods, such as ReduceByKey, which is functionally equivalent to the MR Reducer (see below for further information on what we should really be using, mapPartitions):

val processedRDD = distDataRDD.map(s =>{
   // Processing as before to build the SimpleFeatureType
   (new Text, simpleFeatureType)
})

Then, we can finally output to Accumulo using the GeomesaOutputFormat from the original job:

processedRDD.saveAsNewAPIHadoopFile("output/path", classOf[Text], classOf[SimpleFeatureType], classOf[GeomesaOutputFormat])

At this stage, we have not mentioned the setup method in the MR job; this method is called before any input is processed to allocate an expensive resource like a database connection, or in our case, a reusable object, and a cleanup method is then used to release that resource if it were to persist when out of scope. In our case, the setup method is used to create a SimpleFeatureBuilder which can be reused during each call of the mapper to build SimpleFeatures for output; there is no cleanup method as the memory is automatically released when the object is out of scope (the code has completed).

The Spark map function only operates on one input at a time, and provides no means to execute code before or after transforming a batch of values. It looks reasonable to simply put the setup and cleanup code before and after a call to map:

// do setup work 
val processedRDD = distDataRDD.map(s =>{ 
   // Processing as before to build the SimpleFeatureType 
   (new Text, simpleFeatureType) 
}) 
// do cleanup work 

But, this fails for several reasons:

The closest counterpart to a mapper in Spark is the mapPartitions method. This method does not map just one value to another value, but maps an Iterator of values to an Iterator of other values, akin to a bulk-map method. This means that the mapPartitions can allocate resources locally at its start:

val processedRDD = distDataRDD.mapPartitions { valueIterator =>
   // setup code for SimpleFeatureBuilder
   val transformed = valueIterator.map( . . . )
   transformed
}

However, releasing resources (cleanup) is not straightforward as we still experience the lazy evaluation problem; if resources are freed after the map, then the iterator may not have evaluated before the disappearance of those resources. One solution to this is as follows:

val processedRDD = distDataRDD.mapPartitions { valueIterator =>
  if (valueIterator.isEmpty) {
    // return an Iterator
  } else {
    //  setup code for SimpleFeatureBuilder
    valueIterator.map { s =>
// Processing as before to build the SimpleFeatureType
      val simpleFeature =
      if (!valueIterator.hasNext) {
       // cleanup here
      }
      simpleFeature
    }
  }
}

Now that we have the Spark code for ingest, there is an additional change that we could make, which is to add a Geohash field (see the following for more information on how to produce this field). To insert this field into the code, we will need an additional entry at the end of the GDELT attributes list:

Geohash:String 

And a line to set the value of the simpleFeature type:

simpleFeature.setAttribute(Geomesa, calculatedGeoHash)

Finally, we can run our Spark job to load the GeoMesa Accumulo instance with the GDELT data from HDFS. The two years of GDELT is around 100 million entries! You can check how much data is in Accumulo by using the Accumulo shell, run from the accumulo/bin directory:

./accumulo shell -u username -p password -e "scan -t gdelt_records -np" | wc

Once GeoServer is running, we can create a layer for visualization. GeoServer enables us to publish a single or a group of layers to produce a graphic. When we create a layer, we can specify the bounding box, view the feature (which is the SimpleFeature we created in the Spark code previously), and even run a Common Query Language (CQL) query to filter the data (more about this as follows). After a layer has been created, selecting layer preview and the JPG option will produce a URL with a graphic similar to the following; temporal bounding here is for January 2016 so that the map is not overcrowded:

The URL can be used to produce other graphics, simply by manipulating the arguments. A brief breakdown of the URL is given as follows:

The geoserver URL with the standard:

http://localhost:8080/geoserver/geomesa/wms?

The request type:

service=WMS&version=1.1.0&request=GetMap& 

The layers and styles:

layers=geomesa:event&styles=& 

Set the layer transparency, if required:

transparency=true& 

The cql statement, in this case any row that has an entry with GoldsteinScale>8:

cql_filter=GoldsteinScale>8& 

The bounding box bbox:

bbox=-180.0,-90.0,180.0,90.0& 

The height and width of the graphic:

width=768&height=384& 

Source and image type:

srs=EPSG:4326&format=image%2Fjpeg& 

Filter the content by temporal query bounds:

time=2016-01-01T00:00:00.000Z/2016-01-30T23:00:00.000Z 

The final step for this section is to attach a world map to this layer so that the image becomes more readable. If you search the Internet for world map shape files, there are a number of options; we have used one from http://thematicmapping.org. Adding one of these into GeoServer as a shape-file store, and then creating and publishing a layer before creating a layer group of our GDELT data and the shape-file, will produce an image similar to this:

To make things a bit more interesting, we have filtered the events based on the GoldsteinScale field in the FeatureType. By adding cql_filter=GoldsteinScale > 8 to the URL, we can plot all of the points where the GoldsteinScale score was greater than eight; so essentially, the above image shows us where the highest levels of positive sentiment were located in the world, in January 2016!

Common Query Language (CQL) is a plain text query language created by the OGC for the Catalogue Web Services specification. It is a human-readable query language (unlike, for example, OGC filters) and uses a similar syntax to SQL. Although similar to SQL, CQL has much less functionality; for example, it is quite strict in requiring an attribute to be on the left side of any comparison operator.

The following lists the CQL supported operators:

Due to the limitations of CQL, GeoServer provides an extended version of CQL called ECQL. ECQL provides much of the missing functionality of CQL, providing a more flexible language that has more in common with SQL. GeoServer supports the use of both CQL and ECQL in WMS and WFS requests.

The quickest way to test CQL queries is to amend the URL of a layer such as the one we created above, when using JPGs for example, or to use the CQL box at the bottom of the layer option within GeoMesa.

If we have several layers defined in one WMS request, such as:

http://localhost:8080/geoserver/wms?service=WMS&version=1.1.0&request=GetMap&layers=layer1,layer2,layer3   ...   

Then we may want to filter just one of those layers with the CQL query. In this case, CQL filters must be ordered in the same way that the layers are; we use the INCLUDE keyword for the layers that we don't want to filter and delimit them using a ";". For example, to filter only layer2 in our example, the WMS request would appear thus:

http://localhost:8080/geoserver/wms?service=WMS&version=1.1.0&request=GetMap&layers=layer1,layer2,layer3&cql_filter=INCLUDE;(LAYER2_COL='value');INCLUDE...