Create Object-centric Model from Relational Schema
Jump to navigation
Jump to search
This page describes how to create an object-centric model from a relational data schema. Tables in the relational data are either objects or events in the object-centric model, and relations between the tables are described using foreign keys (which will become the object-to-object and event-to-object relations). The rest of the datatables needed by the object-centric model are created automatically by a script from the object and event tables.
Model creation principle
The object-centric model is built with following principles:
- Each datatable is either an object type or event type datatable.
- For object type tables:
- Each table row is an object.
- OcelObjectTypeObjectId column contains the object ID.
- Each timestamp in the table is an event.
- The event type name is the column name.
- The event is linked to the object where the timestamp is as an object attribute (only one object is linked to each event).
- Foreign keys between the object tables define the object-to-object relations.
- For event type tables:
- Each row is an event
- The left-most date type of column contains the event timestamp.
- OcelEventTypeEventId column contains the event ID.
- Column with a foreign key to an object table's OcelObjectTypeObjectId column is the link to the object.
Model creation steps
To create an object-centric model using this method, proceed as follows:
- Create the object and event tables. Each object table needs to have the column named OcelObjectTypeObjectId containing unique ID's for objects. Each event table needs to have the column named OcelEventTypeEventId containing unique ID's for events.
- Define relations between the object and event tables using foreign keys.
- Create an object-centric model and define datatables for the model. For the Object types and Event types tables, define the tables created previously. In addition, define names for the Objects, Events, ObjectToObject, and EventToObject datatables (these datatables will be created by the script).
- Run the script to create the rest of the object-centric model datatables.
Define foreign keys
Foreign keys define how the object tables (and thus individual objects) are linked together. The object-to-object relations in the object-centric model will be based on them. Foreign keys can be defined using the expression language. Here is an example script that sets some foreign keys:
// Set foreign key from datatable "Sales Order Item" to datatable "Sales Order Header". Source and target datatable column name is "VBELN".
Project.DatatableByName("Sales Order Item").SetForeignKeys([
#{
"Columns": [
#{ "Source": "VBELN", "Target": "VBELN" }
],
"TargetDatatable": "Sales Order Header"
}
]);
// Set foreign key from datatable "Delivery Item" to datatable "Delivery Header". Source and target datatable column name is "VBELN".
// Set foreign key from datatable "Delivery Item" to datatable "Sales Order Item". Source datatable columns are "VGBEL" and "VGPOS", and corresponding columns in the target datatable are "VBELN" and "POSNR".
Project.DatatableByName("Delivery Item").SetForeignKeys([
#{
"Columns": [
#{ "Source": "VBELN", "Target": "VBELN" }
],
"TargetDatatable": "Delivery Header"
},
#{
"Columns": [
#{ "Source": "VGBEL", "Target": "VBELN" },
#{ "Source": "VGPOS", "Target": "POSNR" }
],
"TargetDatatable": "Sales Order Item"
}
]);
Project.DatatableByName("Sales Order Changed").SetForeignKeys([
#{
"Columns": [
#{ "Source": "Sales Order ID", "Target": "OcelObjectTypeObjectId" }
],
"TargetDatatable": "Sales Order Header"
}
]);
Create object-centric model datatables
This script uses the object types and event types datatables defined for the object-centric model as the source data to create the events, objects, event-to-object, and object-to-object tables (if they don't exist). The script can also be used to update existing datatables when the data in the object types datatables changes. The script assumes that there is only one object-centric model in the project where the script is run.
let modelObject = Project.Models.Where(IsOcelModel);
if (Count(modelObject) == 0) {
throw "There is no object-centric model in the project - there should be one model.";
} else if (Count(modelObject) > 1) {
throw "There are several object-centric models in the project - there should be one model.";
}
modelObject = modelObject[0];
let COLUMN_EVENTS_EVENT_ID = "OcelEventId";
let COLUMN_EVENTS_TYPE = "OcelEventType";
let COLUMN_EVENTS_TIME = "OcelEventTime";
let COLUMN_OBJECTS_OBJECT_ID = "OcelObjectId";
let COLUMN_OBJECTS_TYPE = "OcelObjectType";
let COLUMN_EVENT_TYPES_EVENT_ID = "OcelEventTypeEventId";
let COLUMN_OBJECT_TYPES_OBJECT_ID = "OcelObjectTypeObjectId";
let COLUMN_EVENT_OBJECT_EVENT_ID = "OcelEventToObjectSourceId";
let COLUMN_EVENT_OBJECT_OBJECT_ID = "OcelEventToObjectTargetId";
let COLUMN_EVENT_OBJECT_QUALIFIER = "OcelEventToObjectQualifier";
let COLUMN_OBJECT_OBJECT_SOURCE_ID = "OcelObjectToObjectSourceId";
let COLUMN_OBJECT_OBJECT_TARGET_ID = "OcelObjectToObjectTargetId";
let COLUMN_OBJECT_OBJECT_QUALIFIER = "OcelObjectToObjectQualifier";
function getDatatableAndCreateIfNeeded(datatableName) {
let dt = Project.DatatableByName(datatableName);
if (dt == null) {
dt = Project.CreateDatatable(#{
"Name": datatableName,
"Type": "Snowflake"
});
}
return dt;
}
// Add data to Objects table from ObjectTypes tables.
let objectTypeNames = modelObject.Configuration.OcelDataSource.ObjectTypes.Keys;
let allObjectsDataFrame;
for (let i = 0; i < Count(objectTypeNames); i++) {
let objectsDatatableName = modelObject.Configuration.OcelDataSource.ObjectTypes[objectTypeNames[i]];
let objectsDataFrame = getDatatableAndCreateIfNeeded(objectsDatatableName)
.SqlDataFrame
.Select([COLUMN_OBJECT_TYPES_OBJECT_ID])
.RenameColumns([COLUMN_OBJECTS_OBJECT_ID: COLUMN_OBJECT_TYPES_OBJECT_ID])
.WithColumn(COLUMN_OBJECTS_TYPE, #expr{ objectTypeNames[i]})
.Select([COLUMN_OBJECTS_OBJECT_ID, COLUMN_OBJECTS_TYPE]);
if (allObjectsDataFrame == _EMPTY) {
allObjectsDataFrame = objectsDataFrame;
} else {
allObjectsDataFrame = allObjectsDataFrame.Append(objectsDataFrame);
}
}
getDatatableAndCreateIfNeeded(modelObject.Configuration.OcelDataSource.Objects)
.Truncate()
.Import(allObjectsDataFrame, #{ "ImportExistingColumnOnly": false });
// Add data to Events table from ObjectTypes tables.
let allEventsDataFrame;
for (let i = 0; i < Count(objectTypeNames); i++) {
let objectsDatatableName = modelObject.Configuration.OcelDataSource.ObjectTypes[objectTypeNames[i]];
let objectsDatatable = getDatatableAndCreateIfNeeded(objectsDatatableName);
let dateColumns = objectsDatatable.Columns.Where(DataType == "DateTime").Name;
for (let j = 0; j < Count(dateColumns); j++) {
let dateColumn = dateColumns[j];
let eventsDataFrame = objectsDatatable
.SqlDataFrame
.Where(Column(#expr{ dateColumn }) != null)
.Select([COLUMN_OBJECT_TYPES_OBJECT_ID, #expr{ dateColumn }])
.WithColumn(COLUMN_EVENTS_EVENT_ID, Concat(Column(COLUMN_OBJECT_TYPES_OBJECT_ID), "_", #expr{ dateColumn }))
.WithColumn(COLUMN_EVENTS_TYPE, #expr{ dateColumn })
.RenameColumns([COLUMN_EVENTS_TIME: #expr{ dateColumn }])
.WithColumn("Object Type", #expr{ objectTypeNames[i]})
.WithColumn("Object ID", Column(COLUMN_OBJECT_TYPES_OBJECT_ID))
.Select([COLUMN_EVENTS_EVENT_ID, COLUMN_EVENTS_TYPE, COLUMN_EVENTS_TIME, "Object Type", "Object ID"]);
if (allEventsDataFrame == _EMPTY) {
allEventsDataFrame = eventsDataFrame;
} else {
allEventsDataFrame = allEventsDataFrame.Append(eventsDataFrame);
}
}
}
// Add data to Events table from EventTypes tables.
let eventTypeNames = modelObject.Configuration.OcelDataSource.EventTypes.Keys;
for (let i = 0; i < Count(eventTypeNames); i++) {
let eventsDatatableName = modelObject.Configuration.OcelDataSource.EventTypes[eventTypeNames[i]];
let eventsDatatable = getDatatableAndCreateIfNeeded(eventsDatatableName);
let dateColumns = eventsDatatable.Columns.Where(DataType == "DateTime").Name;
if (Count(dateColumns) == 0) {
throw "EventTypes datatable \"" + eventsDatatableName + "\" should have a date type of column for timestamps.";
}
let foreignKeys = eventsDatatable.ForeignKeys;
if (Count(foreignKeys) == 0) {
throw "Foreign key to the ObjectType table is missing from EventTypes datatable \"" + eventsDatatableName + "\".";
}
let objectsTable = foreignKeys[0].TargetDatatable;
let sourceColumn = foreignKeys[0].Columns[0].Source;
let targetColumn = foreignKeys[0].Columns[0].Target;
if (!objectsTable.In(objectTypeNames)) {
throw "Foreign key in the EventTypes datatable \"" + eventsDatatableName + "\" is not refering to an object type table.";
}
if (targetColumn != COLUMN_OBJECT_TYPES_OBJECT_ID) {
throw "Foreign key in the EventTypes datatable \"" + eventsDatatableName + "\" is not refering to an OcelObjectTypeObjectId column.";
}
let eventsDataFrame = eventsDatatable
.SqlDataFrame
.RenameColumns([COLUMN_EVENTS_EVENT_ID: COLUMN_EVENT_TYPES_EVENT_ID])
.WithColumn(COLUMN_EVENTS_TYPE, #expr{ eventsDatatableName })
.WithColumn(COLUMN_EVENTS_TIME, Column(dateColumns[0]))
.WithColumn("Object Type", #expr{ objectsTable })
.WithColumn("Object ID", Column(sourceColumn))
.Select([COLUMN_EVENTS_EVENT_ID, COLUMN_EVENTS_TYPE, COLUMN_EVENTS_TIME, "Object Type", "Object ID"]);
if (allEventsDataFrame == _EMPTY) {
allEventsDataFrame = eventsDataFrame;
} else {
allEventsDataFrame = allEventsDataFrame.Append(eventsDataFrame);
}
}
getDatatableAndCreateIfNeeded(modelObject.Configuration.OcelDataSource.Events)
.Truncate()
.Import(allEventsDataFrame, #{ "ImportExistingColumnOnly": false });
// Add data to Event-to-Object table from ObjectTypes tables.
let allEventToObjectDataFrame;
for (let i = 0; i < Count(objectTypeNames); i++) {
let objectsDatatableName = modelObject.Configuration.OcelDataSource.ObjectTypes[objectTypeNames[i]];
let objectsDatatable = getDatatableAndCreateIfNeeded(objectsDatatableName);
let dateColumns = objectsDatatable.Columns.Where(DataType == "DateTime").Name;
for (let j = 0; j < Count(dateColumns); j++) {
let dateColumn = dateColumns[j];
let eventToObjectDataFrame = objectsDatatable
.SqlDataFrame
.Where(Column(#expr{ dateColumn }) != null)
.Select([COLUMN_OBJECT_TYPES_OBJECT_ID])
.WithColumn(COLUMN_EVENT_OBJECT_EVENT_ID, Concat(Column(COLUMN_OBJECT_TYPES_OBJECT_ID), "_", #expr{ dateColumn }))
.RenameColumns([COLUMN_EVENT_OBJECT_OBJECT_ID: COLUMN_OBJECT_TYPES_OBJECT_ID])
.WithColumn(COLUMN_EVENT_OBJECT_QUALIFIER, "")
.Select([COLUMN_EVENT_OBJECT_EVENT_ID, COLUMN_EVENT_OBJECT_OBJECT_ID, COLUMN_EVENT_OBJECT_QUALIFIER]);
if (allEventToObjectDataFrame == _EMPTY) {
allEventToObjectDataFrame = eventToObjectDataFrame;
} else {
allEventToObjectDataFrame = allEventToObjectDataFrame.Append(eventToObjectDataFrame);
}
}
}
// Add data to Event-to-Object table from EventTypes tables.
for (let i = 0; i < Count(eventTypeNames); i++) {
let eventsDatatableName = modelObject.Configuration.OcelDataSource.EventTypes[eventTypeNames[i]];
let eventsDatatable = getDatatableAndCreateIfNeeded(eventsDatatableName);
let objectIdColumn = eventsDatatable.ForeignKeys[0].Columns[0].Source;
let eventToObjectDataFrame = eventsDatatable
.SqlDataFrame
.RenameColumns([COLUMN_EVENT_OBJECT_EVENT_ID: COLUMN_EVENT_TYPES_EVENT_ID])
.WithColumn(COLUMN_EVENT_OBJECT_OBJECT_ID, Column(objectIdColumn))
.WithColumn(COLUMN_EVENT_OBJECT_QUALIFIER, "")
.Select([COLUMN_EVENT_OBJECT_EVENT_ID, COLUMN_EVENT_OBJECT_OBJECT_ID, COLUMN_EVENT_OBJECT_QUALIFIER]);
if (allEventToObjectDataFrame == _EMPTY) {
allEventToObjectDataFrame = eventToObjectDataFrame;
} else {
allEventToObjectDataFrame = allEventToObjectDataFrame.Append(eventToObjectDataFrame);
}
}
getDatatableAndCreateIfNeeded(modelObject.Configuration.OcelDataSource.EventToObject)
.Truncate()
.Import(allEventToObjectDataFrame, #{ "ImportExistingColumnOnly": false });
// Add data to Object-to-object table from ObjectTypes tables.
let allObjectToObjectDataFrame;
for (let i = 0; i < Count(objectTypeNames); i++) {
let objectsDatatableName = modelObject.Configuration.OcelDataSource.ObjectTypes[objectTypeNames[i]];
let objectsDatatable = getDatatableAndCreateIfNeeded(objectsDatatableName);
let foreignKeys = objectsDatatable.ForeignKeys;
for (let j = 0; j < Count(foreignKeys); j++) {
let joinedColumns = foreignKeys[j].Columns;
let targetDataFrame = Project.DatatableByName(foreignKeys[j].TargetDatatable).SqlDataFrame
.Select(Distinct(Flatten([COLUMN_OBJECT_TYPES_OBJECT_ID, joinedColumns.Target])))
.RenameColumns(Flatten([COLUMN_OBJECT_TYPES_OBJECT_ID, joinedColumns.Target]).{ let col = _;[("Target" + col): col] }[0]);
let objectToObjectDataFrame = objectsDatatable.SqlDataFrame
.Select(Distinct(Flatten([COLUMN_OBJECT_TYPES_OBJECT_ID, joinedColumns.Source])))
.Join(targetDataFrame, joinedColumns.{ let col = _;[col.Source: ("Target" + col.Target)] }[0], "inner")
.WithColumn(COLUMN_OBJECT_OBJECT_SOURCE_ID, Column(COLUMN_OBJECT_TYPES_OBJECT_ID))
.WithColumn(COLUMN_OBJECT_OBJECT_TARGET_ID, Column("TargetOcelObjectTypeObjectId"))
.WithColumn(COLUMN_OBJECT_OBJECT_QUALIFIER, "")
.Select([COLUMN_OBJECT_OBJECT_SOURCE_ID, COLUMN_OBJECT_OBJECT_TARGET_ID, COLUMN_OBJECT_OBJECT_QUALIFIER]);
if (allObjectToObjectDataFrame == _EMPTY) {
allObjectToObjectDataFrame = objectToObjectDataFrame;
} else {
allObjectToObjectDataFrame = allObjectToObjectDataFrame.Append(objectToObjectDataFrame);
}
}
}
getDatatableAndCreateIfNeeded(modelObject.Configuration.OcelDataSource.ObjectToObject)
.Truncate()
.Import(allObjectToObjectDataFrame, #{ "ImportExistingColumnOnly": false });