Загрузка данных из XES

Эта статья демонстрирует загрузку журнала событий, и таблицы процессов из единого набора даных в формате XES.

1. Перейдите к редактированию рабочей области, в которую необходимо загрузить данные. Нажмите на блок Подготовка данных и ML.

   

2. В появившемся интерфейсе Jupyter Lab загрузите необходимые XES-файлы.

   

Результат:

3. Нажмите Новый лаунчер.

   

4. В списке доступных лаунчеров, выберите Amphi→New Pipeline. Это создаст новый пайплайн Amphi.

   

5. В окне редактирования пайплайна, перетащите узел Python Input из панели инструментов на рабочую область.

6. Нажмите , чтобы перейти к настройке узла.

7. В окне настройке узла Python Input введите код, необходимый для загрузки Pandas DataFrame из файла:

    
   import pm4pyoutput = pm4py.read_xes("example-log.xes")

   

8. Нажмите кнопку Input-узла, чтобы проверить корректность загрузки данных. При удачном получении данных, консоль пайплайна отобразит загруженную таблицу (Pandas DataFrame).

   

Для загрузки данных в формате, поддерживаемом VK Business Analytics, необходимо применить ряд трансформаций к исходному DataFrame.

1. Перетащите на рабочую поверхность два узла Python Transfoms. Для удобства, вы можете переименовать эти узлы в Get Cases и Get Events.

2. Добавьте Python-код необходимый для извлечения списка процессов и списка событий в режиме редактирования этих узлов:

Get Cases (список процессов):

 
# --- Transform input to VKPM format ---# Columns that are case-level attributes (prefixed with "case:")case_columns = [c for c in input.columns if c.startswith("case:")]# Columns that are event-level but NOT standard XES fieldsstandard_event_cols = {"concept:name", "time:timestamp", "lifecycle:transition"}event_attr_columns = [c for c in input.columns if c not in standard_event_cols and c not in case_columns]# --- Build cases (processes) dataframe ---case_id_col = "case:concept:name"case_attr_cols = [c for c in case_columns if c != case_id_col]cases_df = input.drop_duplicates(subset=[case_id_col])[[case_id_col] + case_attr_cols].copy()cases_df = cases_df.reset_index(drop=True)cases_df.index = cases_df.index + 1  # 1-based process_id# Build attributes string (key:value pairs)cases_df["attributes"] = cases_df[case_attr_cols].apply(   lambda row: ",".join(f"{col.replace('case:', '')}:{row[col]}" for col in case_attr_cols),   axis=1,)# Map case name -> process_idcase_to_pid = dict(zip(cases_df[case_id_col], cases_df.index))xes_processes = cases_df[["attributes"]].copy()xes_processes.index.name = "process_id"xes_processes = xes_processes.reset_index()output = xes_processes

Get Events (список событий):

 
# --- Build events dataframe ---# Columns that are case-level attributes (prefixed with "case:")case_columns = [c for c in input.columns if c.startswith("case:")]case_id_col = "case:concept:name"case_attr_cols = [c for c in case_columns if c != case_id_col]# Columns that are event-level but NOT standard XES fieldsstandard_event_cols = {"concept:name", "time:timestamp", "lifecycle:transition"}event_attr_columns = [c for c in input.columns if c not in standard_event_cols and c not in case_columns]# Split into start and complete events, then merge to get start/end per activitystarts = input[input["lifecycle:transition"] == "start"].copy()completes = input[input["lifecycle:transition"] == "complete"].copy()# Add a helper key to pair start/complete within the same case and activity occurrencestarts["_case"] = starts[case_id_col]starts["_activity"] = starts["concept:name"]starts["_order"] = starts.groupby(["_case", "_activity"]).cumcount()completes["_case"] = completes[case_id_col]completes["_activity"] = completes["concept:name"]completes["_order"] = completes.groupby(["_case", "_activity"]).cumcount()cases_df = input.drop_duplicates(subset=[case_id_col])[[case_id_col] + case_attr_cols].copy()cases_df = cases_df.reset_index(drop=True)cases_df.index = cases_df.index + 1  # 1-based process_id# Build attributes string (key:value pairs)cases_df["attributes"] = cases_df[case_attr_cols].apply(   lambda row: ",".join(f"{col.replace('case:', '')}:{row[col]}" for col in case_attr_cols),   axis=1,)# Map case name -> process_idcase_to_pid = dict(zip(cases_df[case_id_col], cases_df.index))events_merged = starts.merge(   completes[["_case", "_activity", "_order", "time:timestamp"]],   on=["_case", "_activity", "_order"],   suffixes=("_start", "_end"),)# Build the events dataframeevents_merged["process_id"] = events_merged[case_id_col].map(case_to_pid)events_merged = events_merged.sort_values(["process_id", "time:timestamp_start"]).reset_index(drop=True)events_merged.index = events_merged.index + 1  # 1-based event_idevents_merged["event_id"] = events_merged.indexevents_merged["event_name"] = events_merged["concept:name"]events_merged["start"] = events_merged["time:timestamp_start"].dt.strftime("%Y-%m-%d %H:%M:%S")events_merged["end"] = events_merged["time:timestamp_end"].dt.strftime("%Y-%m-%d %H:%M:%S")events_merged["sorting"] = 0# Build attributes string from event-level attribute columnsevents_merged["attributes"] = events_merged[event_attr_columns].apply(   lambda row: ",".join(f"{col}:{row[col]}" for col in event_attr_columns),   axis=1,)xes_events = events_merged[["process_id", "event_id", "event_name", "start", "end", "sorting", "attributes"]].copy()output = xes_events

3. Подключите входы обоих трансформационных узлов к Input-узлу.

   

4. DataFrame на выходе узла GetEvents содержит колонки start и end c временными штампами в формате text. Для корректного экспорта, необходимо сконвертировать эти колонки в формат Pandas timestamp. Для этого, добавьте дополнительный трансформационный узел после узла GetEvents.

   

Добавьте к узлу следующий код:

 
def convert_columns_to_timestamp(df: pd.DataFrame, columns, \*, errors="coerce", utc=False) -> pd.DataFrame:   result = df.copy()   for col in columns:      result\[col\] = pd.to_datetime(result\[col\], errors=errors, utc=utc)   return resultoutput = convert_columns_to_timestamp(input, \["start", "end"\])

1. В окне редактирования пайплайна, перетащите два узла Database Output из панели инструментов на рабочую область.

   

2. Нажмите , чтобы перейти к настройке узла.

3. В окне настройке узла Database Output, укажите следующие необходимые настройки:

   Опция	Описание
   Host	Имя хоста СУБД
   Port	Номер порта
   Database Name	Имя базы данных, соответствет UUID рабочей области без дефисов и добавленым спереди символом нижнего подчёркивания. Например: — UUID рабочей области: fa776f95-df6c-4118-9d04-5f95fad4b666 — Имя базы данных: _fa776f95df6c41189d045f95fad4b666
   Schema	Схеба БД, по умолчанию public
   Username	Имя пользователя БД
   Password	Пароль
   Database Type	Тип базы данных (выбрать PostgreSQL)
   Table Name	Таблица в которую необходимо добавить загруженные данные. Для создания новой таблицы, раскрыть выпадающий список, ввести имя таблицы в поле Custom и нажать + Add, затем выбрать добавленную таблицу в списке.
   If Table Exists	Что делать, если таблица уже существует (выдать ошибку / пересоздать / дописать в существующую)?
   Mode	Insert

   Укажите разные имена таблиц для соответствующих ветвей пайплайна, например xes_events и xes_processes.

   

4. Соедините входоs Output-узлов с соответствующими ветвями пайплайна:

   

5. Нажмите (Run Pipeline).

   В случае успешной загрузки данных, консоль пайплайна отобразит уведомление.

   

Создайте модель данных для добавленных таблиц и отчёт, привязанный к этой модели.