1466-instantiate-csv-data-as-rdf: added RDF conversion using RML rules and the RML mapper tool

Author: qhouyee

stack-clients/src/main/java/com/cmclinnovations/stack/clients/core/EndpointNames.java

@@ -7,6 +7,7 @@ public final class EndpointNames {
     public static final String GRAFANA = "grafana";
     public static final String GEOSERVER = "geoserver";
     public static final String RML = "yarrrml-parser";
+    public static final String RML_JAVA = "rml-mapper";
 
     private EndpointNames() {
     }

stack-clients/src/main/java/com/cmclinnovations/stack/clients/core/datasets/RML.java

@@ -16,8 +16,9 @@ public boolean usesBlazegraph() {
   void loadInternal(Dataset parent) {
     Path subdirectory = this.getSubdirectory()
         .orElseThrow(() -> new RuntimeException("No 'subdirectory' specified - required for RML data"));
-    Map<String, byte[]> rmlRules = RmlMapperClient.getInstance().parseYarrrmlToRml(
-        parent.getDirectory().resolve(subdirectory), parent.getNamespace());
-    RmlMapperClient.getInstance().parseRmlToRDF(rmlRules);
+    Path fullSubDirPath = parent.getDirectory().resolve(subdirectory);
+    RmlMapperClient mapperClient = RmlMapperClient.getInstance();
+    Map<String, byte[]> rmlRules = mapperClient.parseYarrrmlToRml(fullSubDirPath, parent.getNamespace());
+    mapperClient.parseRmlToRDF(fullSubDirPath, rmlRules);
   }
 }

stack-clients/src/main/java/com/cmclinnovations/stack/clients/rml/RmlMapperClient.java

@@ -9,6 +9,7 @@
 import java.nio.file.Path;
 import java.text.MessageFormat;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -26,7 +27,9 @@ public class RmlMapperClient extends ClientWithEndpoint<RmlMapperEndpointConfig>
 
   private static final String CSV_FILE_EXTENSION = "csv";
   private static final String YML_FILE_EXTENSION = "yml";
+  private static final String TTL_FILE_EXTENSION = "ttl";
   private static final String YARRRML_PARSER_EXECUTABLE_PATH = "/app/bin/parser.js";
+  private static final String TEMP_CONTAINER_DATA_DIR_PATH = "/data";
 
   private static RmlMapperClient instance = null;
 
@@ -57,10 +60,33 @@ public Map<String, byte[]> parseYarrrmlToRml(Path dirPath, String namespace) {
    * Parses the RML rules into RDF triples that will be uploaded at the target
    * namespace.
    * 
+   * @param dirPath  Target directory path.
    * @param rmlRules Input RML rules.
    */
-  public void parseRmlToRDF(Map<String, byte[]> rmlRules) {
-    LOGGER.info("Reading the RML rules...");
+  public void parseRmlToRDF(Path dirPath, Map<String, byte[]> rmlRules) {
+    LOGGER.info("Uploading the RML rules and csv files into the target container...");
+    String rmlMapperJavaContainerId = super.getContainerId(EndpointNames.RML_JAVA);
+
+    List<String> csvFiles = rmlRules.keySet().stream()
+        .map(file -> FileUtils.replaceExtension(file, CSV_FILE_EXTENSION))
+        .collect(Collectors.toList());
+    super.sendFiles(rmlMapperJavaContainerId, dirPath.toAbsolutePath().toString(), csvFiles,
+        TEMP_CONTAINER_DATA_DIR_PATH);
+    super.sendFilesContent(rmlMapperJavaContainerId, rmlRules, TEMP_CONTAINER_DATA_DIR_PATH);
+
+    LOGGER.info("Converting and uploading csv data...");
+    ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+    ByteArrayOutputStream errorStream = new ByteArrayOutputStream();
+    rmlRules.keySet().stream().forEach(file -> {
+      LOGGER.info("Executing RML rules for {}...", file);
+      String execId = super.createComplexCommand(rmlMapperJavaContainerId, "java", "-jar", "/rmlmapper.jar", "-m",
+          "/data/" + file, "-s", "turtle")
+          .withOutputStream(outputStream)
+          .withErrorStream(errorStream)
+          .exec();
+      super.handleErrors(errorStream, execId, LOGGER);
+    });
+    super.deleteDirectory(rmlMapperJavaContainerId, TEMP_CONTAINER_DATA_DIR_PATH);
   }
 
   /**
@@ -118,14 +144,14 @@ private Map<String, byte[]> genRmlRules(Path dirPath, String namespace) {
               }
             }));
     // Send the files to a new data directory
-    super.sendFilesContent(containerId, yarrrmlRules, "/data");
+    super.sendFilesContent(containerId, yarrrmlRules, TEMP_CONTAINER_DATA_DIR_PATH);
 
-    // Execute the command and return the RML rules alongside the CSV file name
+    // Execute the command and return the RML rules alongside the TTL file name
     ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
     ByteArrayOutputStream errorStream = new ByteArrayOutputStream();
     Map<String, byte[]> rmlRules = yarrrmlRules.entrySet().stream()
         .collect(Collectors.toMap(
-            entry -> FileUtils.replaceExtension(entry.getKey(), CSV_FILE_EXTENSION),
+            entry -> FileUtils.replaceExtension(entry.getKey(), TTL_FILE_EXTENSION),
             entry -> {
               LOGGER.info("Generating RML rules from {}...", entry.getKey());
               String execId = super.createComplexCommand(containerId, YARRRML_PARSER_EXECUTABLE_PATH, "-i",
@@ -137,7 +163,7 @@ private Map<String, byte[]> genRmlRules(Path dirPath, String namespace) {
               return outputStream.toByteArray();
             }));
     LOGGER.info("RML rules are generated. Removing any temporary YARRRML files in the container...");
-    super.deleteDirectory(containerId, "/data");
+    super.deleteDirectory(containerId, TEMP_CONTAINER_DATA_DIR_PATH);
     return rmlRules;
   }
 

stack-clients/src/main/java/com/cmclinnovations/stack/clients/rml/RmlMapperJavaEndpointConfig.java

@@ -0,0 +1,41 @@
+package com.cmclinnovations.stack.clients.rml;
+
+import com.cmclinnovations.stack.clients.core.PasswordEndpointConfig;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+
+@JsonInclude(Include.NON_NULL)
+public class RmlMapperJavaEndpointConfig extends PasswordEndpointConfig {
+  private final String hostName;
+  private final String port;
+  private final String username;
+
+  protected RmlMapperJavaEndpointConfig() {
+    this(null, null, null, null, null);
+  }
+
+  public RmlMapperJavaEndpointConfig(String name, String hostName, String port, String username, String passwordFile) {
+    super(name, passwordFile);
+    this.hostName = hostName;
+    this.port = port;
+    this.username = username;
+  }
+
+  public String getHostName() {
+    return hostName;
+  }
+
+  public String getPort() {
+    return port;
+  }
+
+  public String getUsername() {
+    return username;
+  }
+
+  @JsonIgnore
+  public String getServiceUrl() {
+    return "http://" + hostName + ":" + port + "/rml-mapper";
+  }
+}

stack-clients/src/main/java/com/cmclinnovations/stack/clients/utils/YarrrmlFile.java

@@ -128,7 +128,7 @@ private void appendSources(Map<String, Object> output, Path filePath) {
 
         Map<String, Object> sources = new HashMap<>(this.sourcesTemplate);
         Map<String, Object> sourceRef = this.castToMapStringObject(sources.get(SOURCE_REF_KEY));
-        sourceRef.put(ACCESS_KEY, FileUtils.replaceExtension(this.fileName, "csv"));
+        sourceRef.put(ACCESS_KEY, "/data/" + FileUtils.replaceExtension(this.fileName, "csv"));
         output.put(SOURCES_KEY, sources);
     }
 
@@ -170,9 +170,10 @@ private void updateMappings(Map<String, Object> output) {
                 stringObjectMap.put(TARGETS_KEY, TARGET_REF_KEY);
             }
 
-            // Transformation of po if necessary to be YARRRML compliant 
+            // Transformation of po if necessary to be YARRRML compliant
             // SnakeYAML transforms the YML content for nested lists into - -
-            // BUT YARRRML parser only accepts -[] as a shortcut and should be updated accordingly
+            // BUT YARRRML parser only accepts -[] as a shortcut and should be updated
+            // accordingly
             List<Map<String, String>> transformedPo = new ArrayList<>();
             List<Object> originalPo = this.castToListObject(mappingValue.get(PRED_OBJ_KEY));
             for (Object predObj : originalPo) {

stack-clients/src/main/java/com/cmclinnovations/stack/services/RmlMapperJavaService.java

@@ -0,0 +1,12 @@
+package com.cmclinnovations.stack.services;
+
+import com.cmclinnovations.stack.services.config.ServiceConfig;
+
+public class RmlMapperJavaService extends ContainerService {
+
+  public static final String TYPE = "rml-mapper";
+
+  public RmlMapperJavaService(String stackName, ServiceConfig config) {
+    super(stackName, config);
+  }
+}

stack-clients/src/main/resources/com/cmclinnovations/stack/services/built-ins/rml-mapper.json

@@ -0,0 +1,16 @@
+{
+  "type": "rml-mapper",
+  "ServiceSpec": {
+    "Name": "rml-mapper",
+    "TaskTemplate": {
+      "ContainerSpec": {
+        "Image": "rmlio/rmlmapper-java:v7.3.1",
+        "Command": [
+          "tail",
+          "-f",
+          "/dev/null"
+        ]
+      }
+    }
+  }
+}

stack-clients/src/test/java/com/cmclinnovations/stack/clients/utils/YarrrmlFileTest.java

@@ -105,7 +105,7 @@ private String genExpectedYarrrmlContents(String fileName) throws IOException {
         Map<String, Map<String, Object>> sources = new LinkedHashMap<>();
         Map<String, Object> sourceRef = new LinkedHashMap<>();
         sourceRef.put("referenceFormulation", "csv");
-        sourceRef.put("access", FileUtils.replaceExtension(fileName, "csv"));
+        sourceRef.put("access", "/data/" + FileUtils.replaceExtension(fileName, "csv"));
         sources.put("source-ref", sourceRef);
         yamlData.put("sources", sources);