Calculate distance to expert distribution layers

livingatlas/configs/la-pipelines-local-hdfs.yaml

@@ -66,3 +66,7 @@ sample-avro:
 index:
     inputPath: hdfs://localhost:8020/pipelines-data
     zkHost: localhost:9983
+outlier:
+    inputPath: hdfs://localhost:8020/pipelines-data
+    targetPath: hdfs://localhost:8020/pipelines-outlier
+

livingatlas/configs/la-pipelines-local.yaml

@@ -1 +1,4 @@
 # which filesystem to use: local or hdfs
+outlier:
+  baseUrl: https://spatial-test.ala.org.au/ws/
+  runner: DirectRunner

livingatlas/configs/la-pipelines.yaml

@@ -167,6 +167,15 @@ sampling:
   allDatasetsInputPath: '{fsPath}/pipelines-all-datasets'
   runner: SparkRunner
 
+# Calculate distance to the expert distribution layers
+# class:au.org.ala.pipelines.beam.DistributionOutlierPipeline
+outlier:
+  appName: Expert distribution outliers for {datasetId}
+  baseUrl: https://spatial-test.ala.org.au/ws/
+  targetPath: '{fsPath}/pipelines-outlier'
+  allDatasetsInputPath: '{fsPath}/pipelines-all-datasets'
+  runner: SparkRunner
+
 # class: au.org.ala.pipelines.beam.ALAInterpretedToSensitivePipeline
 sensitive:
   appName: Sensitive Data Processing for {datasetId}
@@ -192,6 +201,7 @@ solr:
   allDatasetsInputPath: '{fsPath}/pipelines-all-datasets'
   jackKnifePath: "{fsPath}/pipelines-jackknife"
   clusteringPath: "{fsPath}/pipelines-clustering"
+  outlierPath: '{fsPath}/pipelines-outlier'
   solrCollection: biocache
   includeSampling: false
   includeJackKnife: false
@@ -334,6 +344,18 @@ sampling-sh-args:
     executor-memory: 16G
     driver-memory: 4G
 
+outlier-sh-args:
+  local:
+    jvm: -Xmx8g -XX:+UseG1GC -Dspark.master=local[*]
+  spark-embedded:
+    jvm: -Xmx8g -XX:+UseG1GC -Dspark.master=local[*]
+  spark-cluster:
+    conf: spark.default.parallelism=144
+    num-executors: 8
+    executor-cores: 8
+    executor-memory: 16G
+    driver-memory: 4G
+
 sensitive-sh-args:
   spark-embedded:
     jvm: -Xmx8g -XX:+UseG1GC
@@ -354,7 +376,7 @@ index-sh-args:
   spark-embedded:
     jvm: -Xmx8g -XX:+UseG1GC
   spark-cluster:
-    conf: spark.default.parallelism=500
+    conf: spark.default.parallelism=48
     num-executors: 6
     executor-cores: 8
     executor-memory: 20G

livingatlas/pipelines/src/main/java/au/org/ala/distribution/DistributionLayer.java

@@ -0,0 +1,37 @@
+package au.org.ala.distribution;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.PropertyNamingStrategy;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Represent a field in the sampling service. A field is a property derived from a spatial layer.
+ */
+@Getter
+@Setter
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy.class)
+public class DistributionLayer {
+  private int gid;
+  private String data_resource_uid;
+  private String scientific;
+  private String pid;
+  private String type;
+  private String genus_name;
+  private String lsid;
+  private float area_km;
+  private String common_nam;
+  private int spcode;
+  private String genus_lsid;
+  private String bounding_box;
+  private String group_name;
+  private Boolean endemic;
+  private String image_url;
+  private String family_lsid;
+  private String specific_n;
+  private String wmsurl;
+  private String family;
+  private int geom_idx;
+}

livingatlas/pipelines/src/main/java/au/org/ala/distribution/DistributionService.java

@@ -0,0 +1,32 @@
+package au.org.ala.distribution;
+
+import java.util.List;
+import java.util.Map;
+import retrofit2.Call;
+import retrofit2.http.*;
+
+/** Simple client to the ALA sampling service. */
+public interface DistributionService {
+
+  /** Return expert distribution layers in the ALA spatial portal */
+  @GET("distribution/")
+  Call<List<DistributionLayer>> getLayers();
+
+  /**
+   * @param id the species id MUST be URLEncoded
+   * @param nowkt
+   * @return
+   */
+  @GET("distribution/lsids/{id}")
+  Call<List<DistributionLayer>> getLayersByLsid(
+      @Path(value = "id") String id, @Query("nowkt") String nowkt);
+
+  /**
+   * @param id the species id MUST be URLEncoded
+   * @param points Map<uuid, <decimalLatitude,decimalLongitude>>
+   * @return
+   */
+  @POST("distribution/outliers/{id}")
+  Call<Map<String, Double>> outliers(
+      @Path(value = "id") String id, @Body Map<String, Map<String, Double>> points);
+}

livingatlas/pipelines/src/main/java/au/org/ala/distribution/DistributionServiceImpl.java

@@ -0,0 +1,99 @@
+package au.org.ala.distribution;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategy;
+import java.io.IOException;
+import java.io.Serializable;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import retrofit2.Response;
+import retrofit2.Retrofit;
+import retrofit2.converter.jackson.JacksonConverterFactory;
+
+@Slf4j
+public class DistributionServiceImpl implements Serializable {
+
+  private Retrofit retrofit;
+  private DistributionService service;
+
+  private DistributionServiceImpl(String baseUrl) {
+    ObjectMapper om =
+        new ObjectMapper().setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
+    JacksonConverterFactory jcf = JacksonConverterFactory.create(om);
+    retrofit = new Retrofit.Builder().baseUrl(baseUrl).addConverterFactory(jcf).build();
+    service = retrofit.create(DistributionService.class);
+  }
+
+  /**
+   * Init Distribution outlier service
+   *
+   * @param baseUrl Url of Spatial service
+   * @return
+   */
+  public static DistributionServiceImpl init(String baseUrl) {
+    return new DistributionServiceImpl(baseUrl);
+  }
+
+  public List<DistributionLayer> getLayers() throws IOException, ExpertDistributionException {
+    Response<List<au.org.ala.distribution.DistributionLayer>> response =
+        service.getLayers().execute();
+    int code = response.code();
+    if (code >= 200 && code < 300) {
+      List<au.org.ala.distribution.DistributionLayer> layers = response.body();
+      return layers;
+    } else {
+      errorHandler(code, response);
+      return null;
+    }
+  }
+
+  public List<DistributionLayer> findLayersByLsid(String lsid)
+      throws IOException, ExpertDistributionException {
+    lsid = URLEncoder.encode(lsid, StandardCharsets.UTF_8.toString());
+    Response<List<DistributionLayer>> response = service.getLayersByLsid(lsid, "false").execute();
+    int code = response.code();
+    if (code >= 200 && code < 300) {
+      List<au.org.ala.distribution.DistributionLayer> layers = response.body();
+      return layers;
+    } else {
+      log.error("Error in finding the expert distribution layer of " + lsid);
+      errorHandler(code, response);
+      return null;
+    }
+  }
+
+  public Map<String, Double> outliers(String lsid, Map<String, Map<String, Double>> points)
+      throws IOException, ExpertDistributionException {
+    lsid = URLEncoder.encode(lsid, StandardCharsets.UTF_8.toString());
+    Response<Map<String, Double>> response = service.outliers(lsid, points).execute();
+    int code = response.code();
+    if (code >= 200 && code < 300) {
+      Map<String, Double> results = response.body();
+      return results;
+    } else {
+      errorHandler(code, response);
+      return null;
+    }
+  }
+
+  /**
+   * Todo Handle error information Need to sync with Spatial Service
+   *
+   * @param code
+   * @param response
+   * @throws IOException
+   * @throws ExpertDistributionException
+   */
+  private void errorHandler(int code, Response<?> response)
+      throws IOException, ExpertDistributionException {
+    String errorBody = response.errorBody().string();
+    if (code >= 400 && code < 500) {
+      throw new ExpertDistributionException(errorBody);
+    } else {
+      throw new RuntimeException(errorBody);
+    }
+  }
+}

livingatlas/pipelines/src/main/java/au/org/ala/distribution/ExpertDistributionException.java

@@ -0,0 +1,7 @@
+package au.org.ala.distribution;
+
+public class ExpertDistributionException extends Exception {
+  public ExpertDistributionException(String errorMessage) {
+    super(errorMessage);
+  }
+}