(My apologies if this is a re-post due to email-address issues)
I have experienced problems ingesting any of the sample GDELT data cited from the GeoMesa tutorial instructions on
http://geomesa.github.io/2014/04/17/geomesa-gdelt-analysis/.
Consequently, I tried to adapt the code from https://github.com/ccri/geomesa-gdelt to ingest a different data-set.
My data (in.CSV format—I did convert to .TSV format for ingestion) looks like:
EventID,Sensor,Time,Latitude,Longitude,Altitude,MinAxis,MajAxis,Orientation
1,123,1400086863513,36.966,-82.589,0,792.458,960.222,337.966
2,123,1400086864516,35.783,-82.647,0,229.497,552.263,66.236
3,123,1400086865518,37.215,-84.237,0,283.729,825.958,38.794
4,123,1400086866521,37.609,-82.255,0,737.285,919.335,108.117
5,123,1400086867523,35.053,-82.897,0,710.054,942.109,46.797
6,123,1400086868525,37.145,-84.328,0,695.256,976.911,40.299
7,123,1400086869528,34.224,-84.809,0,645.225,936.959,260.217
8,123,1400086870530,36.826,-85.623,0,430.444,794.505,257.836
9,123,1400086871542,34.846,-86.305,0,727.479,732.795,8.621
10,123,1400086872556,33.861,-86.148,0,700.236,761.757,160.848
. . .
(Note: the “Time” field is a Unix/Java-style time: i.e., # milliseconds since 1970.01.01).
I adapted the GDELTIngest class to SensorIngest.java as shown below.
I also adapted the GDELTIngestMapper class to SensorIngestMapper.java as shown below.
I essentially used the technique from the tutorial to upload my data-file to HDFS (after converting to .TSV format).
I then used basically the same ingestion command from the tutorial:
hadoop jar ./target/geomesa-sensor-1.0-SNAPSHOT.jar geomesa.sensor.SensorIngest -instanceId ntc-irad -zookeepers localhost:2181 -user root -password “…” -auths
public -tableName sensor -featureName sensor -ingestFile hdfs:///sensor/sensor.tsv
The ingestion begins, the “sensor” table is created in Accumulo, but after processing 4 “entries”, the ingestion fails with the following NullPointerException (in Hadoop’s MapReduce?):
ava.lang.Exception: java.lang.NullPointerException
at org.apache.hadoop.mapred.LocalJobRunner$Job.runTasks(LocalJobRunner.java:462)
at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:522)
Caused by: java.lang.NullPointerException
at geomesa.core.data.AccumuloFeatureReader.<init>(AccumuloFeatureReader.scala:42)
at geomesa.core.data.AccumuloDataStore.getFeatureReader(AccumuloDataStore.scala:283)
at geomesa.core.data.AccumuloDataStore.getFeatureReader(AccumuloDataStore.scala:275)
at geomesa.core.data.ModifyAccumuloFeatureWriter.<init>(AccumuloFeatureWriter.scala:146)
at geomesa.core.data.AccumuloDataStore.createFeatureWriter(AccumuloDataStore.scala:294)
at org.geotools.data.AbstractDataStore.getFeatureWriter(AbstractDataStore.java:470)
at org.geotools.data.AbstractDataStore.getFeatureWriter(AbstractDataStore.java:503)
at geomesa.sensor.SensorIngestMapper.setup(SensorIngestMapper.java:64)
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:142)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:340)
at org.apache.hadoop.mapred.LocalJobRunner$Job$MapTaskRunnable.run(LocalJobRunner.java:243)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:744)
14/05/14 13:44:37 INFO mapreduce.Job: Job job_local1747290386_0001 failed with state FAILED due to: NA
14/05/14 13:44:37 INFO mapreduce.Job: Counters: 0
Exception in thread "main" java.lang.Exception: Job failed
at geomesa.sensor.SensorIngest.runMapReduceJob(SensorIngest.java:156)
at geomesa.sensor.SensorIngest.main(SensorIngest.java:112)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.hadoop.util.RunJar.main(RunJar.java:212)
The line tagged with “!!!” in SensorIngestMapper.java listing is where the code is dying (in the DataStore.getFeatureWriter call).
I’m probably missing something simple, but so far I’ve been unable to get this simple data-file to ingest into Accumulo using the GeoMesa/Hadoop mechanism (although I can ingest
data directly into Accumulo).
Also, I’m not sure what attribute-type to associate with the “Time” field in the “attributes” List. I’ve tried “:Time”, “:Timestamp”, “:Datetime”, etc. with no success, so I’ve
had to label it simply as “:Integer”.
Any help or suggestions would be greatly appreciated!
Regards,
Bob Barnhart (barnhartr@xxxxxxxxxx)
// SensorIngest.java ===========================================================================================
package geomesa.sensor;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import geomesa.core.index.Constants;
import org.apache.accumulo.core.client.mapreduce.AccumuloFileOutputFormat;
import org.apache.commons.cli.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.geotools.data.DataStore;
import org.geotools.data.DataStoreFinder;
import org.geotools.data.DataUtilities;
import org.geotools.feature.SchemaException;
import org.opengis.feature.simple.SimpleFeatureType;
import java.util.*;
public class SensorIngest {
static String INSTANCE_ID = "instanceId";
static String ZOOKEEPERS = "zookeepers";
static String USER = "user";
static String PASSWORD = "password";
static String AUTHS = "auths";
static String TABLE_NAME = "tableName";
static String FEATURE_NAME = "featureName";
static String INGEST_FILE = "ingestFile";
static String[] ACCUMULO_CONNECTION_PARAMS = new String[]{INSTANCE_ID,
ZOOKEEPERS, USER, PASSWORD, AUTHS, TABLE_NAME};
static Options getCommonRequiredOptions() {
Options options = new Options();
Option instanceIdOpt = OptionBuilder.withArgName(INSTANCE_ID)
.hasArg()
.isRequired()
.withDescription("accumulo connection parameter instanceId")
.create(INSTANCE_ID);
Option zookeepersOpt = OptionBuilder.withArgName(ZOOKEEPERS)
.hasArg()
.isRequired()
.withDescription("accumulo connection parameter zookeepers")
.create(ZOOKEEPERS);
Option userOpt = OptionBuilder.withArgName(USER)
.hasArg()
.isRequired()
.withDescription("accumulo connection parameter user")
.create(USER);
Option passwordOpt = OptionBuilder.withArgName(PASSWORD)
.hasArg()
.isRequired()
.withDescription("accumulo connection parameter password")
.create(PASSWORD);
Option authsOpt = OptionBuilder.withArgName(AUTHS)
.hasArg()
.withDescription("accumulo connection parameter auths")
.create(AUTHS);
Option tableNameOpt = OptionBuilder.withArgName(TABLE_NAME)
.hasArg()
.isRequired()
.withDescription("accumulo connection parameter tableName")
.create(TABLE_NAME);
Option featureNameOpt = OptionBuilder.withArgName(FEATURE_NAME)
.hasArg()
.isRequired()
.withDescription("name of feature in accumulo table")
.create(FEATURE_NAME);
options.addOption(instanceIdOpt);
options.addOption(zookeepersOpt);
options.addOption(userOpt);
options.addOption(passwordOpt);
options.addOption(authsOpt);
options.addOption(tableNameOpt);
options.addOption(featureNameOpt);
return options;
}
public static Map<String, String> getAccumuloDataStoreConf(CommandLine cmd) {
Map<String , String> dsConf = new HashMap<String , String>();
for (String param : ACCUMULO_CONNECTION_PARAMS) {
dsConf.put(param, cmd.getOptionValue(param));
}
if (dsConf.get(AUTHS) == null) dsConf.put(AUTHS, "");
return dsConf;
}
public static void main(String [ ] args) throws Exception {
CommandLineParser parser = new BasicParser();
Options options = getCommonRequiredOptions();
Option ingestFileOpt = OptionBuilder.withArgName(INGEST_FILE)
.hasArg()
.isRequired()
.withDescription("ingest tsv file on hdfs")
.create(INGEST_FILE);
options.addOption(ingestFileOpt);
CommandLine cmd = parser.parse( options, args);
Map<String, String> dsConf = getAccumuloDataStoreConf(cmd);
String featureName = cmd.getOptionValue(FEATURE_NAME); // e.g. "sensor"
SimpleFeatureType featureType = buildFeatureType(featureName);
DataStore ds = DataStoreFinder.getDataStore(dsConf);
ds.createSchema(featureType);
runMapReduceJob(featureName,
"GeoMesa SensorEvent Ingest",
SensorIngestMapper.class,
dsConf,
new Path(cmd.getOptionValue(INGEST_FILE)));
}
private static void runMapReduceJob(String featureName, String jobName, Class mapper,
Map<String, String> dsConf,
Path mapredCSVFilePath) throws Exception {
Job job = Job.getInstance(new Configuration());
job.setMapperClass(mapper);
job.setOutputFormatClass(AccumuloFileOutputFormat.class);
job.setJobName(jobName);
Configuration conf = job.getConfiguration();
for (String key : dsConf.keySet()) {
conf.set(key, dsConf.get(key));
}
conf.set(FEATURE_NAME, featureName);
FileSystem fs = FileSystem.get(conf);
FileInputFormat.setInputPaths(job, mapredCSVFilePath);
Path tmpPath = new Path("///tmp");
if (!fs.exists(tmpPath)) {
fs.create(tmpPath);
}
Path outputDir = new Path("///tmp", "geomesa-"+featureName+"-output");
if (fs.exists(outputDir)) {
// remove this directory, if it already exists
fs.delete(outputDir, true);
}
Path hdfsJarPath = new Path("///tmp", "geomesa-"+featureName+"-1.0-SNAPSHOT.jar");
if (fs.exists(hdfsJarPath)) {
// remove this jar, if it already exists
fs.delete(hdfsJarPath, true);
}
FileOutputFormat.setOutputPath(job, outputDir);
fs.copyFromLocalFile(new Path("target/geomesa-"+featureName+"-1.0-SNAPSHOT.jar"), hdfsJarPath);
for (FileStatus path : fs.listStatus(hdfsJarPath)) {
job.addArchiveToClassPath(new Path(path.getPath().toUri().getPath()));
}
job.submit();
if (!job.waitForCompletion(true)) {
throw new Exception("Job failed");
}
}
private static SimpleFeatureType buildFeatureType(String featureName) throws SchemaException {
String name = featureName;
String spec = Joiner.on(",").join(attributes);
SimpleFeatureType featureType = DataUtilities.createType(name, spec);
featureType.getUserData().put(Constants.SF_PROPERTY_START_TIME, "Time"); // Tells GeoMesa to use "Time" as the Start Time index
return featureType;
}
static List<String> attributes = Lists.newArrayList(
"EventID:Integer",
"Sensor:Integer",
"Time:Integer", // TODO: ISSUE I001 - actually a "Unix" time (e.g., millis since 1970.01.01)
"Latitude:Float",
"Longitude:Float",
"Altitude:Integer",
"MinAxis:Float",
"MajAxis:Float",
"Orientation:Float"
);
}
// SensorIngestMapper.java ====================================================================================
package geomesa.sensor;
import com.vividsolutions.jts.geom.Coordinate;
import com.vividsolutions.jts.geom.Geometry;
import com.vividsolutions.jts.geom.GeometryFactory;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.geotools.data.DataStore;
import org.geotools.data.DataStoreFinder;
import org.geotools.data.FeatureWriter;
import org.geotools.data.Transaction;
import org.geotools.feature.simple.SimpleFeatureBuilder;
import org.geotools.filter.identity.FeatureIdImpl;
import org.geotools.geometry.jts.JTSFactoryFinder;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.feature.simple.SimpleFeatureType;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.Map;
public class SensorIngestMapper extends Mapper<LongWritable,Text,Key,Value> {
// Columns: EventID, Sensor, Time, Latitude, Longitude, Altitude, MinAxis, MajAxis, Orientation
private static int ID_COL_IDX = 0;
private static int TIME_COL_IDX = 2;
private static int LATITUDE_COL_IDX = 3;
private static int LONGITUDE_COL_IDX = 4;
private static int MINIMUM_NUM_FIELDS = 9;
private SimpleFeatureType featureType = null;
private FeatureWriter<SimpleFeatureType, SimpleFeature> featureWriter = null;
private SimpleFeatureBuilder featureBuilder;
private GeometryFactory geometryFactory = JTSFactoryFinder.getGeometryFactory();
private static void _(String s) {
System.out.println("("+s+")");
}
public void setup(Mapper<LongWritable,Text,Key,Value>.Context context) throws IOException, InterruptedException {
super.setup(context);
_("SensorIngestMapper.setup: ENTER...");
Map<String , String> connectionParams = new HashMap<String , String>();
connectionParams.put("instanceId", context.getConfiguration().get("instanceId"));
connectionParams.put("zookeepers", context.getConfiguration().get("zookeepers"));
connectionParams.put("user", context.getConfiguration().get("user"));
connectionParams.put("password", context.getConfiguration().get("password"));
connectionParams.put("auths", context.getConfiguration().get("auths"));
connectionParams.put("tableName", context.getConfiguration().get("tableName"));
String featureName = context.getConfiguration().get("featureName");
_("SensorIngestMapper.setup: featureName="+featureName);
DataStore ds = DataStoreFinder.getDataStore(connectionParams);
featureType = ds.getSchema(featureName);
_("SensorIngestMapper.setup: featureType="+featureType.getTypeName());
featureBuilder = new SimpleFeatureBuilder(featureType);
_("SensorIngestMapper.setup: featureBuilder OK...");
featureWriter = ds.getFeatureWriter(featureName, Transaction.AUTO_COMMIT); // NullPointerException on this call!!!
_("SensorIngestMapper.setup: RETURN.");
}
public void map(LongWritable key, Text value, Mapper<LongWritable,Text,Key,Value>.Context context) {
_("SensorIngestMapper.map: ENTER...");
String[] attributes = value.toString().split("\\t", -1);
if (attributes.length >= MINIMUM_NUM_FIELDS && !attributes[LATITUDE_COL_IDX].equals("") && !attributes[LONGITUDE_COL_IDX].equals("")) {
SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd");
try {
featureBuilder.reset();
featureBuilder.addAll(attributes);
Double lat = Double.parseDouble(attributes[LATITUDE_COL_IDX]); // TODO: update for Lat/Lon offsets
Double lon = Double.parseDouble(attributes[LONGITUDE_COL_IDX]);
Geometry geom = geometryFactory.createPoint(new Coordinate(lon, lat));
SimpleFeature simpleFeature = featureBuilder.buildFeature(attributes[ID_COL_IDX]);
simpleFeature.setAttribute("Time", formatter.parse(attributes[TIME_COL_IDX]));
simpleFeature.setDefaultGeometry(geom);
try {
SimpleFeature next = featureWriter.next();
for (int i = 0; i < simpleFeature.getAttributeCount(); i++) {
next.setAttribute(i, simpleFeature.getAttribute(i));
}
((FeatureIdImpl)next.getIdentifier()).setID(simpleFeature.getID());
featureWriter.write();
} catch (IOException e) {
e.printStackTrace();
}
} catch (ParseException e) {
e.printStackTrace();
}
}
_("SensorIngestMapper.map: RETURN ...");
}
}
Bob Barnhart
Chief Systems Engineer | 858 826 5596 (Office) | 619 972 9489 (Mobile) |
barnhartr@xxxxxxxxxx
Bob Barnhart
Chief Systems Engineer | 858 826 5596 (Office) | 619 972 9489 (Mobile) |
Robert.M.Barnhart@xxxxxxxxxx
