bench_executor.rmlmapper

  1#!/usr/bin/env python3
  2
  3"""
  4The RMLMapper executes RML rules to generate high quality Linked Data
  5from multiple originally (semi-)structured data sources.
  6
  7**Website**: https://rml.io<br>
  8**Repository**: https://github.com/RMLio/rmlmapper-java
  9"""
 10
 11import os
 12import psutil
 13from typing import Optional
 14from timeout_decorator import timeout, TimeoutError  # type: ignore
 15from bench_executor.container import Container
 16from bench_executor.logger import Logger
 17
 18VERSION = '6.0.0'
 19TIMEOUT = 6 * 3600  # 6 hours
 20
 21
 22class RMLMapper(Container):
 23    """RMLMapper container for executing R2RML and RML mappings."""
 24
 25    def __init__(self, data_path: str, config_path: str, directory: str,
 26                 verbose: bool):
 27        """Creates an instance of the RMLMapper class.
 28
 29        Parameters
 30        ----------
 31        data_path : str
 32            Path to the data directory of the case.
 33        config_path : str
 34            Path to the config directory of the case.
 35        directory : str
 36            Path to the directory to store logs.
 37        verbose : bool
 38            Enable verbose logs.
 39        """
 40        self._data_path = os.path.abspath(data_path)
 41        self._config_path = os.path.abspath(config_path)
 42        self._logger = Logger(__name__, directory, verbose)
 43        self._verbose = verbose
 44
 45        os.makedirs(os.path.join(self._data_path, 'rmlmapper'), exist_ok=True)
 46        super().__init__(f'kgconstruct/rmlmapper:v{VERSION}', 'RMLMapper',
 47                         self._logger,
 48                         volumes=[f'{self._data_path}/rmlmapper:/data',
 49                                  f'{self._data_path}/shared:/data/shared'])
 50
 51    @property
 52    def root_mount_directory(self) -> str:
 53        """Subdirectory in the root directory of the case for RMLMapper.
 54
 55        Returns
 56        -------
 57        subdirectory : str
 58            Subdirectory of the root directory for RMLMapper.
 59
 60        """
 61        return __name__.lower()
 62
 63    @timeout(TIMEOUT)
 64    def _execute_with_timeout(self, arguments: list) -> bool:
 65        """Execute a mapping with a provided timeout.
 66
 67        Returns
 68        -------
 69        success : bool
 70            Whether the execution was successfull or not.
 71        """
 72        if self._verbose:
 73            arguments.append('-vvvvvvvvvvv')
 74
 75        self._logger.info(f'Executing RMLMapper with arguments '
 76                          f'{" ".join(arguments)}')
 77
 78        # Set Java heap to 1/2 of available memory instead of the default 1/4
 79        max_heap = int(psutil.virtual_memory().total * (1/2))
 80
 81        # Execute command
 82        cmd = f'java -Xmx{max_heap} -Xms{max_heap} ' + \
 83              '-jar rmlmapper/rmlmapper.jar ' + \
 84              f'{" ".join(arguments)}'
 85        return self.run_and_wait_for_exit(cmd)
 86
 87    def execute(self, arguments: list) -> bool:
 88        """Execute RMLMapper with given arguments.
 89
 90        Parameters
 91        ----------
 92        arguments : list
 93            Arguments to supply to RMLMapper.
 94
 95        Returns
 96        -------
 97        success : bool
 98            Whether the execution succeeded or not.
 99        """
100        try:
101            return self._execute_with_timeout(arguments)
102        except TimeoutError:
103            msg = f'Timeout ({TIMEOUT}s) reached for RMLMapper'
104            self._logger.warning(msg)
105
106        return False
107
108    def execute_mapping(self,
109                        mapping_file: str,
110                        output_file: str,
111                        serialization: str,
112                        rdb_username: Optional[str] = None,
113                        rdb_password: Optional[str] = None,
114                        rdb_host: Optional[str] = None,
115                        rdb_port: Optional[int] = None,
116                        rdb_name: Optional[str] = None,
117                        rdb_type: Optional[str] = None) -> bool:
118        """Execute a mapping file with RMLMapper.
119
120        N-Quads and N-Triples are currently supported as serialization
121        format for RMLMapper.
122
123        Parameters
124        ----------
125        mapping_file : str
126            Path to the mapping file to execute.
127        output_file : str
128            Name of the output file to store the triples in.
129        serialization : str
130            Serialization format to use.
131        rdb_username : Optional[str]
132            Username for the database, required when a database is used as
133            source.
134        rdb_password : Optional[str]
135            Password for the database, required when a database is used as
136            source.
137        rdb_host : Optional[str]
138            Hostname for the database, required when a database is used as
139            source.
140        rdb_port : Optional[int]
141            Port for the database, required when a database is used as source.
142        rdb_name : Optional[str]
143            Database name for the database, required when a database is used as
144            source.
145        rdb_type : Optional[str]
146            Database type, required when a database is used as source.
147
148        Returns
149        -------
150        success : bool
151            Whether the execution was successfull or not.
152        """
153        arguments = ['-m', os.path.join('/data/shared/', mapping_file),
154                     '-s', serialization,
155                     '-o', os.path.join('/data/shared/', output_file),
156                     '-d']  # Enable duplicate removal
157
158        if rdb_username is not None and rdb_password is not None \
159                and rdb_host is not None and rdb_port is not None \
160                and rdb_name is not None and rdb_type is not None:
161
162            arguments.append('-u')
163            arguments.append(rdb_username)
164            arguments.append('-p')
165            arguments.append(rdb_password)
166
167            parameters = ''
168            if rdb_type == 'MySQL':
169                protocol = 'jdbc:mysql'
170                parameters = '?allowPublicKeyRetrieval=true&useSSL=false'
171            elif rdb_type == 'PostgreSQL':
172                protocol = 'jdbc:postgresql'
173            else:
174                raise ValueError(f'Unknown RDB type: "{rdb_type}"')
175            rdb_dsn = f'{protocol}://{rdb_host}:{rdb_port}/' + \
176                      f'{rdb_name}{parameters}'
177            arguments.append('-dsn')
178            arguments.append(rdb_dsn)
179
180        return self.execute(arguments)