bench_executor.sdmrdfizer

  1#!/usr/bin/env python3
  2
  3"""
  4SDM-RDFizer is an efficient RML-Compliant engine for knowledge graph
  5construction.
  6
  7**Repository**: https://github.com/SDM-TIB/SDM-RDFizer
  8"""
  9
 10import os
 11import configparser
 12from rdflib import Graph, BNode, Namespace, Literal, RDF
 13from timeout_decorator import timeout, TimeoutError  # type: ignore
 14from typing import Optional
 15from bench_executor.container import Container
 16from bench_executor.logger import Logger
 17
 18VERSION = '4.6.6.5'
 19TIMEOUT = 6 * 3600  # 6 hours
 20R2RML = Namespace('http://www.w3.org/ns/r2rml#')
 21RML = Namespace('http://semweb.mmlab.be/ns/rml#')
 22D2RQ = Namespace('http://www.wiwiss.fu-berlin.de/suhl/bizer/D2RQ/0.1#')
 23IMAGE = f'blindreviewing/sdm-rdfizer:v{VERSION}'
 24
 25
 26class SDMRDFizer(Container):
 27    """SDMRDFizer container for executing RML mappings."""
 28
 29    def __init__(self, data_path: str, config_path: str, directory: str,
 30                 verbose: bool):
 31        """Creates an instance of the SDMRDFizer class.
 32
 33        Parameters
 34        ----------
 35        data_path : str
 36            Path to the data directory of the case.
 37        config_path : str
 38            Path to the config directory of the case.
 39        directory : str
 40            Path to the directory to store logs.
 41        verbose : bool
 42            Enable verbose logs.
 43        """
 44        self._data_path = os.path.abspath(data_path)
 45        self._config_path = os.path.abspath(config_path)
 46        self._logger = Logger(__name__, directory, verbose)
 47
 48        os.umask(0)
 49        os.makedirs(os.path.join(self._data_path, 'sdmrdfizer'), exist_ok=True)
 50        super().__init__(IMAGE, 'SDM-RDFizer', self._logger,
 51                         volumes=[f'{self._data_path}/sdmrdfizer:/data',
 52                                  f'{self._data_path}/shared:/data/shared'])
 53
 54    @property
 55    def root_mount_directory(self) -> str:
 56        """Subdirectory in the root directory of the case for SDM-RDFizer.
 57
 58        Returns
 59        -------
 60        subdirectory : str
 61            Subdirectory of the root directory for SDM-RDFizer.
 62
 63        """
 64        return __name__.lower()
 65
 66    @timeout(TIMEOUT)
 67    def _execute_with_timeout(self, arguments) -> bool:
 68        """Execute a mapping with a provided timeout.
 69
 70        Returns
 71        -------
 72        success : bool
 73            Whether the execution was successfull or not.
 74        """
 75        cmd = 'python3 sdm-rdfizer/rdfizer/run_rdfizer.py ' + \
 76              '/data/config_sdmrdfizer.ini'
 77        return self.run_and_wait_for_exit(cmd)
 78
 79    def execute(self, arguments: list) -> bool:
 80        """Execute SDM-RDFizer with given arguments.
 81
 82        Parameters
 83        ----------
 84        arguments : list
 85            Arguments to supply to SDM-RDFizer.
 86
 87        Returns
 88        -------
 89        success : bool
 90            Whether the execution succeeded or not.
 91        """
 92        try:
 93            return self._execute_with_timeout(arguments)
 94        except TimeoutError:
 95            msg = f'Timeout ({TIMEOUT}s) reached for SDM-RDFizer'
 96            self._logger.error(msg)
 97
 98        return False
 99
100    def execute_mapping(self,
101                        mapping_file: str,
102                        output_file: str,
103                        serialization: str,
104                        rdb_username: Optional[str] = None,
105                        rdb_password: Optional[str] = None,
106                        rdb_host: Optional[str] = None,
107                        rdb_port: Optional[int] = None,
108                        rdb_name: Optional[str] = None,
109                        rdb_type: Optional[str] = None) -> bool:
110        """Execute a mapping file with SDM-RDFizer.
111
112        N-Quads and N-Triples are currently supported as serialization
113        format for RMLMapper.
114
115        Parameters
116        ----------
117        mapping_file : str
118            Path to the mapping file to execute.
119        output_file : str
120            Name of the output file to store the triples in.
121        serialization : str
122            Serialization format to use.
123        rdb_username : Optional[str]
124            Username for the database, required when a database is used as
125            source.
126        rdb_password : Optional[str]
127            Password for the database, required when a database is used as
128            source.
129        rdb_host : Optional[str]
130            Hostname for the database, required when a database is used as
131            source.
132        rdb_port : Optional[int]
133            Port for the database, required when a database is used as source.
134        rdb_name : Optional[str]
135            Database name for the database, required when a database is used as
136            source.
137        rdb_type : Optional[str]
138            Database type, required when a database is used as source.
139
140        Returns
141        -------
142        success : bool
143            Whether the execution was successfull or not.
144        """
145
146        # Configuration file
147        name = os.path.splitext(os.path.basename(output_file))[0]
148        config = configparser.ConfigParser(delimiters=':')
149        config['default'] = {
150            'main_directory': '/data/shared'
151        }
152        config['datasets'] = {
153            'number_of_datasets': str(1),
154            'output_folder': '/data/shared',
155            'all_in_one_file': 'yes',
156            'remove_duplicate': 'yes',
157            'enrichment': 'yes',
158            'name': name,
159            'ordered': 'no',
160            'large_file': 'false'
161        }
162        config['dataset1'] = {
163            'name': name,
164            'mapping': f'/data/shared/{os.path.basename(mapping_file)}'
165        }
166
167        if serialization == 'ntriples':
168            config['datasets']['output_format'] = 'n-triples'
169        elif serialization == 'turtle':
170            config['datasets']['output_format'] = 'turtle'
171        else:
172            raise NotImplementedError('SDM-RDFizer does not support'
173                                      '"serialization" output format')
174
175        if rdb_username is not None and rdb_password is not None \
176                and rdb_host is not None and rdb_port is not None \
177                and rdb_name is not None and rdb_type is not None:
178            config['dataset1']['user'] = rdb_username
179            config['dataset1']['password'] = rdb_password
180            config['dataset1']['host'] = rdb_host
181            config['dataset1']['port'] = str(rdb_port)
182            config['dataset1']['db'] = rdb_name
183            config['dataset1']['mapping'] = '/data/mapping_converted.rml.ttl'
184            if rdb_type == 'MySQL':
185                config['datasets']['dbType'] = 'mysql'
186                driver = 'jdbc:mysql'
187            elif rdb_type == 'PostgreSQL':
188                config['datasets']['dbType'] = 'postgres'
189                driver = 'jdbc:postgresql'
190            else:
191                raise NotImplementedError('SDM-RDFizer does not support RDB '
192                                          f'"{rdb_type}"')
193            dsn = f'{driver}://{rdb_host}:{rdb_port}/{rdb_name}'
194
195            # Compatibility with R2RML mapping files
196            # Replace rr:logicalTable with rml:logicalSource + D2RQ description
197            # and rr:column with rml:reference
198            g = Graph()
199            g.bind('rr', R2RML)
200            g.bind('rml', RML)
201            g.bind('d2rq', D2RQ)
202            g.bind('rdf', RDF)
203            g.parse(os.path.join(self._data_path, 'shared',
204                                 os.path.basename(mapping_file)))
205
206            # rr:logicalTable --> rml:logicalSource
207            for triples_map_iri, p, o in g.triples((None, RDF.type,
208                                                    R2RML.TriplesMap)):
209                logical_source_iri = BNode()
210                d2rq_rdb_iri = BNode()
211                logical_table_iri = g.value(triples_map_iri,
212                                            R2RML.logicalTable)
213                table_name_literal = g.value(logical_table_iri,
214                                             R2RML.tableName)
215                g.add((d2rq_rdb_iri, D2RQ.jdbcDSN, Literal(dsn)))
216                g.add((d2rq_rdb_iri, D2RQ.jdbcDriver, Literal(driver)))
217                g.add((d2rq_rdb_iri, D2RQ.username, Literal(rdb_username)))
218                g.add((d2rq_rdb_iri, D2RQ.password, Literal(rdb_password)))
219                g.add((d2rq_rdb_iri, RDF.type, D2RQ.Database))
220                g.add((logical_source_iri, R2RML.sqlVersion, R2RML.SQL2008))
221                g.add((logical_source_iri, R2RML.tableName,
222                       table_name_literal))
223                g.add((logical_source_iri, RML.source, d2rq_rdb_iri))
224                g.add((logical_source_iri, RDF.type, RML.LogicalSource))
225                g.add((triples_map_iri, RML.logicalSource, logical_source_iri))
226                g.remove((triples_map_iri, R2RML.logicalTable,
227                          logical_table_iri))
228                g.remove((logical_table_iri, R2RML.tableName,
229                          table_name_literal))
230                g.remove((logical_table_iri, RDF.type, R2RML.LogicalTable))
231                g.remove((logical_table_iri, R2RML.sqlVersion, R2RML.SQL2008))
232
233            # rr:column --> rml:reference
234            for s, p, o in g.triples((None, R2RML.column, None)):
235                g.add((s, RML.reference, o))
236                g.remove((s, p, o))
237
238            # SDM-RDFizer cannot handle rml:referenceFormulation when using
239            # RDBs, remove it for safety
240            # https://github.com/SDM-TIB/SDM-RDFizer/issues/71
241            for s, p, o in g.triples((None, RML.referenceFormulation, None)):
242                g.remove((s, p, o))
243
244            destination = os.path.join(self._data_path, 'sdmrdfizer',
245                                       'mapping_converted.rml.ttl')
246            g.serialize(destination=destination, format='turtle')
247
248        os.umask(0)
249        os.makedirs(os.path.join(self._data_path, 'sdmrdfizer'), exist_ok=True)
250        path = os.path.join(self._data_path, 'sdmrdfizer',
251                            'config_sdmrdfizer.ini')
252        with open(path, 'w') as f:
253            config.write(f, space_around_delimiters=False)
254
255        return self.execute([])