fix for mapreduce ingest (#1321)

locationtech · Apr 10, 2018 · 4f01bf3 · 4f01bf3
1 parent e1bddaf
commit 4f01bf3
Showing 1 changed file with 13 additions and 7 deletions.
diff --git a/...rc/main/java/mil/nga/giat/geowave/core/ingest/hdfs/mapreduce/AbstractMapReduceIngest.java b/...rc/main/java/mil/nga/giat/geowave/core/ingest/hdfs/mapreduce/AbstractMapReduceIngest.java
@@ -156,15 +156,21 @@ public int run(
 		GeoWaveOutputFormat.setStoreOptions(
 				job.getConfiguration(),
 				dataStoreOptions);
-		PrimaryIndex[] indexesArray = new PrimaryIndex[indexes.size()];
 		final WritableDataAdapter<?>[] dataAdapters = ingestPlugin.getDataAdapters(ingestOptions.getVisibility());
-		for (final WritableDataAdapter<?> dataAdapter : dataAdapters) {
-			dataAdapter.init(indexes.toArray(indexesArray));
-			GeoWaveOutputFormat.addDataAdapter(
-					job.getConfiguration(),
-					dataAdapter);
+		if (dataAdapters != null && dataAdapters.length > 0) {
+			PrimaryIndex[] indexesArray = indexes.toArray(new PrimaryIndex[indexes.size()]);
+			for (final WritableDataAdapter<?> dataAdapter : dataAdapters) {
+				// from a controlled client, intialize the writer within the
+				// context of the datastore before distributing ingest
+				dataStoreOptions.createDataStore().createWriter(
+						dataAdapter,
+						indexesArray).close();
+
+				GeoWaveOutputFormat.addDataAdapter(
+						job.getConfiguration(),
+						dataAdapter);
+			}
 		}
-
 		job.setSpeculativeExecution(false);
 
 		// add required indices