Process Mining

v2023.4

偽

Process Mining

最終更新日 2024年4月19日

変換を編集する

フォルダー構造

プロセスアプリの変換は、1 つの dbt プロジェクトで成り立ちます。以下の表は、dbt プロジェクトフォルダーの内容の説明です。

フォルダー/ファイル	次の値を含む
`dbt_packages\`	`pm_utils` パッケージとそのマクロです。
`logs\`	dbt の実行時に作成されるログです。
`macros\`	カスタムマクロです。
`models\`	`.sql` 変換を定義するファイル
`models\schema\`	`.yml` データに対するテストを定義するファイル
`seed`	`.csv` 構成設定を含むファイル
`dbt_project.yml`	dbt プロジェクトの設定です。

以下の画像でご確認ください。

データ変換

データ変換は、models\ ディレクトリ内の .sql ファイルで定義されます。データ変換は、サブディレクトリの標準セット内で整理されています。

1_input
2_entities
3_events
4_event_logs
5_business_logic です。

「変換の構造」をご覧ください。

.sql ファイルは Jinja SQL で記述されており、プレーンな SQL クエリに Jinja ステートメントを挿入できます。dbt ですべての .sql ファイルが実行されると、各 .sql ファイルによってデータベース内に新しいビューまたはテーブルが生成されます。

.sql ファイルは通常、次の構造を持ちます。

With ステートメント: 必要なサブテーブルを含む With ステートメントが 1 つ以上必要です。
- {{ ref(‘My_table) }} は、別の .sql ファイルで定義されたテーブルを参照します。
- {{ source(var("schema_sources"), 'My_table') }} は入力テーブルを参照します。
メインクエリ: 新しいテーブルを定義するクエリです。
最後のクエリ: 通常、Select * from table のようなクエリが最後に使用されます。これにより、デバッグ中に副選択を簡単に行うことができます。

変換の効果的な記述方法について詳しくは、「SQL を記述するためのヒント」をご覧ください。

ソーステーブルを追加する

dbt プロジェクトに新しいソーステーブルを追加するには、そのテーブルが models\schema\sources.yml のリストに含められている必要があります。こうすることで、他のモデルが {{ source(var("schema_sources"), 'My_table_raw') }} を使用してテーブルを参照できます。以下の画像で例をご確認ください。

重要: 新しいソーステーブルは、それぞれ sources.yml のリストに加えられている必要があります。

注:

データの読み込み時に、ソーステーブルのテーブル名にサフィックス _raw が追加されます。たとえば、my_table というテーブルは my_table_raw として参照されます。

詳しくは、ソースに関する dbt の公式ドキュメントをご覧ください。

データ出力

データ変換は、対応するアプリで必要となるデータモデルを出力する必要があります。つまり、期待されるそれぞれのテーブルとフィールドが存在する必要があります。

これは実際の場面では、models\5_business_logic 内のテーブルを削除すべきではないということを意味します。また、対応するクエリの出力フィールドは削除すべきではありません。

プロセスアプリに新しいフィールドを追加する場合は、プロセスアプリで利用可能なカスタムフィールドを使用できます。変換のフィールドをカスタムフィールドにマッピングして、出力で利用できるようにします。出力内のカスタムフィールドの名前は、プロセスアプリのデータモデルで記述された名前と一致させてください。

ヒント:

dbt docs コマンドを使用して dbt プロジェクトのドキュメントサイトを生成し、既定のブラウザーで開くことができます。このドキュメントサイトにはリネージグラフも含まれます。リネージグラフには、プロジェクト内の各データテーブル間の関係をグラフィカルに表したエンティティリレーションシップダイアグラムが表示されます。

詳しくは、dbt docs に関する dbt の公式ドキュメントをご覧ください。

マクロ

マクロを使用すると、一般的な SQL の構造を簡単に再利用できます。詳しくは、Jinja マクロの dbt 公式ドキュメントをご覧ください。

pm_utils

pm-utils パッケージには、Process Mining の変換で通常使用されるマクロが一式含まれています。pm_utils マクロについて詳しくは、「ProcessMining-pm-utils」をご覧ください。

以下に pm_utils.optional() マクロを呼び出す Jinja のコードの例を示します。

シード

シードとは、変換にデータテーブルを追加するために使用される csv ファイルです。詳しくは、Jinja のシードに関する dbt の公式ドキュメントをご覧ください。

Process Mining では、変換内のマッピングを簡単に設定できるよう、一般的にシードが使用されます。

シードファイルを編集しても、ファイルはデータベース内で即座に更新されません。シードファイルの新しい内容をデータベースに読み込むよう dbt に指示するには、以下のいずれかを実行します。

dbt seed - シードファイルのテーブルのみ更新します。
dbt build - すべてのモデルとテストも実行します。

注: 最初にシードファイルにデータレコードが含まれていない場合、データベース内のデータ型が正しく設定されていない可能性があります。これを修正するには、run dbt seed --full-refresh を呼び出します。これにより、データベース内の列のセットも更新されます。

アクティビティの設定

activity_configuration.csv ファイルは、アクティビティに関連する追加フィールドを設定するために使用されます。activity_order は、同じタイムスタンプで 2 つのイベントが発生する際のタイブレーカーとして使用されます。以下の画像で例をご確認ください。

テスト

models\schema\ フォルダーには、テストを定義する.yml ファイル一式が含まれます。これらのファイルは、期待されるデータの構造と内容を検証します。詳しくは、テストに関する dbt の公式ドキュメントをご覧ください。

変換を Process Mining で実行すると、各データ取り込みでは sources.yml 内のテストのみが実行されます。これは、入力データの書式が正しいかどうかを確認するために行われます。

注: 変換を編集する際は、テストを適宜更新してください。テストは必要に応じて削除できます。

カスタムのスループット時間メトリック

はじめに

データ変換とダッシュボードの編集をカスタマイズすることで、カスタムのスループット時間メトリックを作成して使用できます。スループット時間は、2 つのアクティビティ A と B の間のタイミングです。以下に、変換の編集時にカスタムのスループット時間メトリックを作成する場合に実行が必要な手順と、スループット時間メトリックをプロセスアプリのダッシュボードで有効化する方法について説明します。

変換を編集してカスタムのスループット時間メトリックを作成する

まず、スループット時間を再計算してから、それを Case フィールドとして利用できるようにする必要があります。

スループット時間を計算する

ケースごとにアクティビティ A とアクティビティ B の間のスループット時間を計算できます。アクティビティは 1 つのケースで複数回発生する可能性があるため、アクティビティの最初の出現を使用するか、それとも最後の出現を使用するかを考慮する必要があります。

イベントログに基づいて追加のモデルベースを作成し、目的のスループット時間を計算します。たとえば、Cases_with_throughput_times です。
このモデルでは、前処理テーブルを作成し、そこで計算に使用するイベント終了を定義します。テーブルごとに、ケース ID とアクティビティのイベント終了が必要です。以下の例に、あるケースでアクティビティ A の最後の出現を選択する方法を示します。
```
Event_end_activity_A as (
    select
        Event_log."Case_ID",
        max(Event_log."Event_end") as "Event_end_activity_A"
    from Event_log
    where Event_log."Activity" = 'Activity A'
    group by Event_log."Case_ID")Event_end_activity_A as (
    select
        Event_log."Case_ID",
        max(Event_log."Event_end") as "Event_end_activity_A"
    from Event_log
    where Event_log."Activity" = 'Activity A'
    group by Event_log."Case_ID")
```
注:
この例で、アクティビティの最初の出現箇所を選択する場合は、max を min に置き換えます。

前処理テーブルをイベントログに結合して実際のスループット時間を計算し、スループット時間テーブルを定義します。

ヒント:

pm-utils パッケージで提供されている datediff 関数を使用して、2 つのイベント終了間の時間差を計算できます。

スループット時間は、アクティビティ A がアクティビティ B の前にあるインスタンスを対象として、ミリ秒単位で計算する必要があります。ミリ秒は、アプリテンプレートで継続時間を定義するために使用する時間の単位です。スループット時間は、既に前処理テーブルのケースごとにグループ化されているため、任意のレコードを選択できます。上の例では、集計の分が使用されています。スループット時間とケース ID を選択するスループット時間テーブルは、以下に示すように定義できます。

Cases_with_throughput_times as (
    select
        Event_log."Case_ID",
        case
            when min(Event_end_activity_A."Event_end_activity_A") <= min(Event_end_activity_B."Event_end_activity_B")
                then {{ pm_utils.datediff('millisecond',
                'min(Event_end_activity_A."Event_end_activity_A")',
                'min(Event_end_activity_B."Event_end_activity_B")') }}
        end as "Throughput_time_activity_A_to_activity_B"
    from Event_log
    left join Event_end_activity_A
        on Event_log."Case_ID" = Event_end_activity_A."Case_ID"
    left join Event_end_activity_B
        on Event_log."Case_ID" = Event_end_activity_B."Case_ID"
    group by Event_log."Case_ID)"Cases_with_throughput_times as (
    select
        Event_log."Case_ID",
        case
            when min(Event_end_activity_A."Event_end_activity_A") <= min(Event_end_activity_B."Event_end_activity_B")
                then {{ pm_utils.datediff('millisecond',
                'min(Event_end_activity_A."Event_end_activity_A")',
                'min(Event_end_activity_B."Event_end_activity_B")') }}
        end as "Throughput_time_activity_A_to_activity_B"
    from Event_log
    left join Event_end_activity_A
        on Event_log."Case_ID" = Event_end_activity_A."Case_ID"
    left join Event_end_activity_B
        on Event_log."Case_ID" = Event_end_activity_B."Case_ID"
    group by Event_log."Case_ID)"

週末を除く日数でスループット時間を計算する

pm-utils パッケージで提供されている date_from_timestamp 関数を使用して、2 つのアクティビティ間の日数を計算できます。さらに diff_weekdays 関数を使用することで週末を除外できます。

アクティビティ A とアクティビティ B の間の平日の日数を計算する方法について、以下のコードの例をご覧ください。

with Event_log as (
    select * from {{ ref('Event_log') }}
),

Activity_A as (
    select
        Event_log."Case_ID",
        min({{ pm_utils.date_from_timestamp('Event_log."Event_end"') }}) as "Date_activity_A"
    from Event_log
    where Event_log."Activity" = 'Receive invoice'
    group by Event_log."Case_ID"
),

Activity_B as (
    select
        Event_log."Case_ID",
        min({{ pm_utils.date_from_timestamp('Event_log."Event_end"') }}) as "Date_activity_B"
    from Event_log
    where Event_log."Activity" = 'Pay invoice'
    group by Event_log."Case_ID"
),

Total_days_minus_weekends as (
    select
        Activity_A."Case_ID",
        Activity_A."Date_activity_A",
        Activity_B."Date_activity_B",
        {{ pm_utils.diff_weekdays('Activity_A."Date_activity_A"', 'Activity_B."Date_activity_B"') }}
    -- Only compute for cases where both dates are known.
    from Activity_A
    inner join Activity_B
        on Activity_A."Case_ID" = Activity_B."Case_ID"
)

select * from Total_days_minus_weekendswith Event_log as (
    select * from {{ ref('Event_log') }}
),

Activity_A as (
    select
        Event_log."Case_ID",
        min({{ pm_utils.date_from_timestamp('Event_log."Event_end"') }}) as "Date_activity_A"
    from Event_log
    where Event_log."Activity" = 'Receive invoice'
    group by Event_log."Case_ID"
),

Activity_B as (
    select
        Event_log."Case_ID",
        min({{ pm_utils.date_from_timestamp('Event_log."Event_end"') }}) as "Date_activity_B"
    from Event_log
    where Event_log."Activity" = 'Pay invoice'
    group by Event_log."Case_ID"
),

Total_days_minus_weekends as (
    select
        Activity_A."Case_ID",
        Activity_A."Date_activity_A",
        Activity_B."Date_activity_B",
        {{ pm_utils.diff_weekdays('Activity_A."Date_activity_A"', 'Activity_B."Date_activity_B"') }}
    -- Only compute for cases where both dates are known.
    from Activity_A
    inner join Activity_B
        on Activity_A."Case_ID" = Activity_B."Case_ID"
)

select * from Total_days_minus_weekends

休日を除く日数でスループット時間を計算する

アクティビティ A とアクティビティ B の間のスループット時間を、週末と休日を除いた日数で計算する手順は以下のとおりです。

1. 休日としてカウントする日を定義する Holidays.csv ファイルを作成します。ファイルには、少なくとも各休日に対して 1 つのレコードが含まれる必要があります。次の形式を使用します。

休日	日付	平日
元日	2024-01-01	はい
イースター	2024-03-31	いいえ
..	..	..

Holidays.csv ファイル内のレコードは、日付範囲から除外する必要がある日数をカウントするために使用されます。

2. Holidays.csv ファイルをシードファイルとして dbt プロジェクトに読み込みます。詳しくは、Jinja のシードに関する dbt の公式ドキュメントをご覧ください。

3. 「週末を除く日数でスループット時間を計算する」で前述したように、pm-utils パッケージで提供される date_from_timestamp と diff_weekdays の関数を使用し、スループット時間を週末を除いた日数で計算します。

4. 休日の .csv ファイルに保存されたレコードのうち、各ケースの指定した日付範囲内にあるレコードの数を計算します。以下の例のコードをご覧ください。

Holidays_count as (
    select
        Total_days_minus_weekends."Case_ID",
        count(Holidays."Date") as "Number_of_holidays"
    from Total_days_minus_weekends
    left join Holidays
        on Holidays."Date" between Total_days_minus_weekends."Date_activity_A" and Total_days_minus_weekends."Date_activity_B"
    where Holidays."Weekday" = 'Yes'
    group by Total_days_minus_weekends."Case_ID"
)Holidays_count as (
    select
        Total_days_minus_weekends."Case_ID",
        count(Holidays."Date") as "Number_of_holidays"
    from Total_days_minus_weekends
    left join Holidays
        on Holidays."Date" between Total_days_minus_weekends."Date_activity_A" and Total_days_minus_weekends."Date_activity_B"
    where Holidays."Weekday" = 'Yes'
    group by Total_days_minus_weekends."Case_ID"
)

注:

上の例では、休日が土曜日または日曜日の場合に休日を減算しないように、フィルター Weekday = 'Yes' が使用されています。これは、diff_weekday 関数ですでに考慮されています。

5. 各ケースで計算された合計日数から、計算された休日数を減算します。以下の例のコードをご覧ください。

Total_days_minus_weekends_and_holidays as (
    select
        Total_days_minus_weekends."Case_ID",
        Total_days_minus_weekends."Number_of_days" - Holidays_count."Number_of_holidays" as "Number_of_days_between_dates"
    from Total_days_minus_weekends
    inner join Holidays_count
        on Total_days_minus_weekends."Case_ID" = Holidays_count."Case_ID"
)Total_days_minus_weekends_and_holidays as (
    select
        Total_days_minus_weekends."Case_ID",
        Total_days_minus_weekends."Number_of_days" - Holidays_count."Number_of_holidays" as "Number_of_days_between_dates"
    from Total_days_minus_weekends
    inner join Holidays_count
        on Total_days_minus_weekends."Case_ID" = Holidays_count."Case_ID"
)

スループット時間をケースフィールドとして利用可能にする

スループット時間テーブルを作成したら、このテーブルを Cases テーブルに結合し、追加のスループット時間データをケースの情報として追加する必要があります。新しいスループット時間フィールドをダッシュボードで利用できるようにするには、新しいスループット時間フィールドをカスタムのケース期間フィールドの 1 つにキャストする必要があります。

Cases テーブル内のカスタムケース期間の行のいずれかを置き換えます。これは次のような行です。

{{ pm_utils.optional(ref('Cases_base'), '"custom_case_duration_1"', 'integer') }} as "custom_case_duration_1",{{ pm_utils.optional(ref('Cases_base'), '"custom_case_duration_1"', 'integer') }} as "custom_case_duration_1",

これを新しく作成したスループット時間に置き換えます。

Cases_with_throughput_times."Throughput_time_activity_A_to_activity_B" as "custom_case_duration_1",Cases_with_throughput_times."Throughput_time_activity_A_to_activity_B" as "custom_case_duration_1",

カスタムのスループット時間メトリックの変換が更新され、アプリテンプレートにインポートできるようになります。

スループット時間メトリックをプロセスアプリのダッシュボードで有効化する

変換でカスタムのスループット時間を作成した場合、そのスループット時間は、アプリテンプレート内のエイリアスでケースのプロパティとして利用可能です。プロセスアプリをカスタマイズすることで、変換で作成したカスタムスループット時間に基づいてスループット時間メトリックを作成できます。

注:

既定では、新しいカスタム期間フィールドが numeric 型のフィールドとして追加されます。必ずフィールドを編集し、新しいフィールドの型を duration に変更してください。「データマネージャー」もご覧ください。

データマネージャーに移動し、新しいメトリックを作成します。
スループット時間に使用するカスタムの期間フィールドを選択し、[平均] または必要な他の集計を選択します。カスタムのケース期間フィールドの名前をデータマネージャーで好きな名前に変更することもできます。
アプリケーションを編集し、新しいメトリックをビジネスユーザーに対して利用可能にするグラフにメトリックを配置します。
ダッシュボードをパブリッシュして、スループット時間メトリックをダッシュボードで利用できるようにします。

注:

Purchase-to-Pay アプリテンプレートと Order-to-Cash アプリテンプレートでは、スループット時間の計算はそれぞれ Purchase_order_items_with_throughput_times と Sales_order_items_with_throughput_times で既に利用可能です。カスタムスループット時間をそこに追加して、Purchase_order_items または Sales_order_items でカスタム期間として利用可能にすることができます。

Snowflake と SQL Server の SQL の違い

SQL Server と Snowflake

ローカルの開発環境では SQL Server で変換が実行され、Automation Suite の Process Mining では Snowflake が使用されます。ほとんどの SQL ステートメントは SQL Server と Snowflake の両方で動作しますが、構文が若干異なる可能性があり、それによって戻り値が異なる場合があります。

両方のデータベースシステムで動作する SQL ステートメントを書くには、以下のようにします。

フィールド名を二重引用符で囲みます (例: Table."Field")。
Snowflake と SQL Server で異なる SQL 関数を使用しないようにします (例:string_agg() と listagg())。

pm_utils パッケージには、両方のデータベースで動作する関数が一式付属しています。詳しくは、「Multiple databases」をご覧ください。たとえば string_agg() や listagg() を使用する代わりに pm_utils.string_agg() を使用すれば、両方のデータベースで結果の挙動が同じになります。pm_utils に目的の関数が含まれていない場合は、各データベースで適切な関数が呼び出されるように Jinja ステートメントを作成する必要があります。

文字列の連結

文字列と結合するには、pm_utils.concat() 関数を使用します。この関数を使用すると、SQL Server と Snowflake の両方で同じ結果が生成されます。

例: pm_utils.concat("This is a nice string", null) = "This is a nice string"文字列の連結は、+ や || などの演算子で行うべきではありません。これらの演算子は両方のデータベースで異なる (Snowflake では || が使用され、SQL Server では + が使用される) ためです。また、標準の concat() 関数の挙動も両方のシステムで異なります。具体的には以下のような違いがあります。

SQL Server	Snowflake
`null` 値は無視され、空の文字列として扱われます。	`null` 値があると、結果全体が `null` になります。

並べ替え

並べ替えは、Snowflake と SQL Server で異なる方法で処理されます。

例： ... order by "Attribute_1" desc, "Attribute_2" ...

null 値

SQL Server	Snowflake
`null` は既定で先頭に並べ替えられます (昇順)。	`null` は既定で末尾に並べ替えられます (昇順)。

大文字の処理

SQL Server	Snowflake
大文字は期待どおりに並べ替えられます (AaBbCc)。	最初に大文字で並べ替えられた後、小文字で並べ替えられます (ABCabc)。

ダッシュ

例： -Accountant-

SQL Server	Snowflake
ダッシュは並べ替え時に無視されます (「-Accountant-」は「Accountant」と同じものとして扱われます)。	ダッシュは先頭に並べ替えられます。

空白文字の処理

値 "A" と " A" をグループ化すると、SQL Server では 1 つの値と見なされますが、Snowflake では 2 つの異なる値として見なされます。したがって、お使いのデータでこの問題が発生する可能性がある場合は、トリミングをお勧めします。

大文字と小文字を区別

既定では、SQL Server では大文字と小文字が区別されません。一方 Snowflake では大文字と小文字が区別されます。つまり、Table."Field" = "Some_value" と Table."Field" = "SOME_VALUE" は、SQL Server では同じ結果のセットが返されますが、Snowflake では 2 つの異なる結果のセットが返される可能性があります。

問題が発生しないように、ローカルの SQL Server データベースの挙動を Snowflake の挙動と一致するように変更することをお勧めします。これは、データベースの照合順序を大文字と小文字を区別する値に設定することで達成できます。

入力データを読み込むための設定

settings.json

The settings.json file contains settings related to loading input data. These settings are temporary and should be used to switch existing process apps (created before March 2023) to the new data loading behavior. You should not change the settings.json file for new process apps.

新しいプロセスアプリを作成する際は、入力データが新しいアプリの作成に使用するアプリテンプレートで求められる形式に沿っていることを必ず確認してください。詳しくは、「アプリテンプレート」をご覧ください。

重要:

新しいプロセスアプリは、既定で新しいデータモデルに適用されます。既存のプロセスアプリは、Process Mining 2023.4 と Process Mining 2023.10 で引き続き動作します。データのアップロードが正しく機能し続けるよう、Process Mining 2024.4 より前のプロセスアプリのデータ読み込み設定を切り替えてください。AddRawTablePostfix および StripSpecialCharacters は一時的な設定であり、Process Mining 20234.4 で削除される予定です。その後は、プロセスアプリはすべて、これらの設定が false に設定されているかのように機能します。

設定	形式	説明
AddRawTablePostfix	Boolean	[データをアップロード] オプションによるファイルのアップロードを使用する際に、ソーステーブルにサフィックス _raw を追加するために使用します。たとえば、アップロードするファイルの名前が Event_log.csv の場合、この設定を true に設定すると、ファイル名が Event_log_raw.csv に変更されます。
StripSpecialCharacters	Boolean	特殊文字を削除し、表名および/またはフィールド名のアンダースコアをスペースで置き換えるために使用します。たとえば、Event+end というフィールドは Eventend に変わります。

既存のプロセスアプリを新しいデータモデル設定で使用する

既存のプロセスアプリを新しいデータモデル設定で使用するには、以下の手順に従います。

settings.json.zip をダウンロードし、解凍します。
プロセスアプリの変換をエクスポートします。「ローカル環境でデータ変換を編集する」をご覧ください。
Add the settings.json file (if not already present) to transformations.
AddRawTablePostfix と StripSpecialCharacters が両方とも false に設定されていることを確認します。
入力ファイルまたは変換を、ファイル名が完全に一致するように変更します。たとえば、入力変換で event_log_raw が予期されている場合は、csv ファイルの名前も event_log_raw.csv にする必要があります。
テーブル名またはフィールド名に特殊文字 (「」「(」「?」など) を使用している場合は、入力変換でも同じ名前を使用してください。たとえば、Activity category というフィールドをクエリで参照する場合は、Activity_category ではなく Activity category を使用する必要があります。
変換をインポートし、新しいデータを取り込みます。

変換がプラットフォームで再度実行され、ソーステーブルが適宜変更されます。

dbt プロジェクト

データ変換は、入力データを Process Mining に適したデータに変換するために使用されます。Process Mining の変換は dbt プロジェクトとして書き込まれます。

このページでは、dbt について説明します。詳しくは、dbt の公式ドキュメントをご覧ください。

pm-utils パッケージ

Process Mining のアプリテンプレートには、pm_utils という dbt パッケージが付属しています。この pm-utils パッケージには、Process Mining の dbt プロジェクト用のユーティリティ関数とマクロが含まれています。pm_utils について詳しくは、「ProcessMining-pm-utils」をご覧ください。