invenio-rdm-migrator 1.0.0a7

Development

pip install -e .

./run-tests.sh

How to run it

To run the migration you need:

• A running InvenioRDM instance.

• If your data contains references to other records (e.g. vocabularies), then it is also required to run the setup step.

invenio-cli services setup --force --no-demo-data

• Install Invenio-RDM-Migrator, any other dependencies must be handled in the Pipfile of your instance.

$ pip install invenio-rdm-migrator

• Create/edit the configuration file on your instance, for example streams.yaml:

records:
    extract:
        filename: /path/to/records.json
    load:
        db_uri: postgresql+psycopg2://inveniordm:inveniordm@localhost:5432/inveniordm
        tmp_dir: /tmp/migrator

• You will need to create a small python script putting together the different blocks of the ETL. You can find an eample at my-site/site/my_site/migrator/__main__.py.

from invenio_rdm_migrator.streams import StreamDefinition
from invenio_rdm_migrator.streams.records import RDMRecordCopyLoad

if __name__ == "__main__":
    RecordStreamDefinition = StreamDefinition(
        name="records",
        extract_cls=JSONLExtract,
        transform_cls=ZenodoToRDMRecordTransform,
        load_cls=RDMRecordCopyLoad,
    )

    runner = Runner(
        stream_definitions=[
            RecordStreamDefinition,
        ],
        config_filepath="path/to/your/streams.yaml",
    )
    runner.run()

• Finally, you can execute the above code. Since it is in the __main__ file of the python package, you can run it as a module:

$ python -m my_site.migrator

• Once the migration has completed, in your instance you can reindex the data. For example, for users and records it would look like:

$ invenio-cli pyshell

In [1]: from invenio_access.permissions import system_identity
In [2]: from invenio_rdm_records.proxies import current_rdm_records_service
In [3]: from invenio_users_resources.proxies import current_users_service

In [4]: current_users_service.rebuild_index(identity=system_identity)
In [5]: current_rdm_records_service.rebuild_index(identity=system_identity)

Implement your {Extract/Transform/Load}

There are for packages in this module extract, transform, load, and streams. The first three correspond to the three steps of an ETL process. The streams package contains the logic to run the process and different stream-specific implementations of ETL classes (e.g. records).

class XMLExtract(Extract):
...

    def run(self):
        with open("file.xml") as file:
            for entry in file:
                yield xml.loads(entry)

{
    "record": self._record(entry),
    "draft": self._draft(entry),
    "parent": self._parent(entry),
}

$ /path/to/data/tables1668697280.943311
    |
    | - pidstore_pid.csv
    | - rdm_parents_metadata.csv
    | - rdm_records_metadata.csv
    | - rdm_versions_state.csv

Notes

Shared cache between streams

During a migration run, there is a need to share information across different streams, e.g populate communities before records and keep the map between community slug names and autogenerated community ids, or at the same stream across different TableGenerator instances, e.g on records stream we keep the “seen” parent ids so we can update the information of the parent for different record versions. For that reason, we pass a cache dict, that can change in the future in a type of persistent storage e.g redis, in each stream.load step so streams can populate/consume the cache. The cache for each stream can also be populated in each stream configuration like below in your streams.yaml:

records:
    extract:
        filename: /path/to/records.json
    load:
        cache:
            communities:
                community_slug: <community_id>
        db_uri: postgresql+psycopg2://inveniordm:inveniordm@localhost:5432/inveniordm
        tmp_dir: /tmp/migrator

When the runner will instantiate each stream will merge the existing state of the cache with whatever is provided in the stream configuration. That means, that the stream configuration takes precedence and can override the whole cache before the stream runs! Any cache state that exists before is overridden for the rest of the migration run.

Infrastructure

While now we are chaining the iterator from one step into the other in the streams, the idea is that all three steps will pull/push to/from queues so they can be deployed in different parts of the system (e.g. the load part in the worker nodes).

Others

• Using generators instead of lists, allows us to iterate through the data only once and perform the E-T-L steps on them. Instead of loop for E, loop for T, loop for L. In addition, this allows us to have the csv files open during the writing and closing them at the end (open/close is an expensive op when done 3M times).

Changes

Version 1.0.0

• Initial public release.