Funnel バッチパイプライン

Funnel バッチパイプラインは、Foundry データソースとユーザーによる編集の両方から取得したデータを、バッチ方式で OSv2 に効率的にインデックス化する処理を統括する内部ジョブパイプラインです。これにより、オントロジーのデータとメタデータを最新の状態に保ちます。

Funnel バッチパイプラインの構成要素

Funnel バッチパイプラインは、一連の Foundry ビルドジョブで構成されます。

• 変更履歴
• 変更のマージ
• インデックス化
• ハイドレーション

以下のスクリーンショットは、Funnel バッチパイプラインの例を示しています。

変更履歴

変更履歴ジョブでは、データソースが新しいデータまたはトランザクションを受け取ると、Funnel がすべてのデータソースのデータ差分を自動的に計算し、Funnel パイプライン内に中間の変更履歴データセットを作成します。変更履歴データセットは、各トランザクションのデータ差分を含む APPEND トランザクションを受け取ることで、インクリメンタル（差分処理）計算のセマンティクスを実現します。これらの変更履歴データセットは Funnel が所有および管理するため、ユーザーはアクセスできません。

変更のマージ

変更のマージジョブでは、変更履歴ステップのすべての変更履歴データセットと、Actions からの直近のユーザー編集を、オブジェクトタイプの主キーで結合します。これにより、すべての変更をマージし、別のデータセットに保存します。これらのマージ済みデータセットは Funnel が所有および管理するため、ユーザーはアクセスできません。

インデックス化

変更がマージされると、Funnel はオブジェクトデータベースごとにインデックス化ジョブを開始し、マージされたすべての変更を含む最終データセットのすべての行を、オブジェクトタイプに設定されたオブジェクトデータベースに対応する形式に変換します。たとえば、標準の OSv2 データベースの場合、前のステップで変更をマージしたデータセットのすべての行がインデックスファイルに変換され、これらのファイルが別のインデックスデータセットに保存されます。これらのインデックスデータセットは Funnel が所有および管理するため、ユーザーはアクセスできません。

ハイドレーション

インデックス化ジョブが完了すると、オブジェクトデータベースはインデックス化済みのデータをクエリできる状態にする必要があります。OSv2 を例にすると、この準備ステップでは、データセットから OSv2 データベースの検索ノードのディスクにインデックスファイルをダウンロードします。ハイドレーションと呼ばれるこのプロセスは、この例の Funnel バッチパイプラインでオブジェクトタイプのデータを更新するための最終ステップです。

ハイドレーションジョブの進行状況は、以下のスクリーンショットに示すように、Ontology Manager アプリケーションに表示されます。

これらのステップが完了すると、オブジェクトタイプは使用可能になり、外部または Foundry 内のほかのサービスからクエリできるようになります。

ライブおよび置き換え用 Funnel パイプライン

オブジェクトタイプのデータまたはスキーマが更新されるときには、2つの独立した Funnel パイプラインが関わります。以下のスクリーンショットは、これら2つの Funnel パイプラインを示しています。

ライブパイプライン

Funnel のライブパイプラインは、Foundry データソースからの新しいデータで、本番環境のオブジェクトタイプを更新します。ライブパイプラインは、対応するデータソースが更新されるたびに実行されます。また、オブジェクトに対するユーザー編集が検出された場合、元データセットの明示的な更新の有無にかかわらず、ライブパイプラインは6時間ごとに実行されます。これにより、インデックス化の変更のマージステップで、ユーザー編集が Funnel 所有のデータセットに保存されます。

ユーザー編集は、オブジェクトデータベースのインデックスに即座に適用されます。6時間ごとにジョブを定期実行することで、組み込みの制御メカニズムがこのデータを Foundry に永続的に保存できます。

置き換え用パイプライン

オブジェクトタイプのスキーマが変更され、以前のパイプラインのスキーマが最新ではなくなると、オブジェクトタイプの更新を統括するために、新しい置き換え用パイプラインをプロビジョニングする必要があります。スキーマの変更には、オブジェクトタイプへの新しいプロパティタイプの追加、既存のプロパティタイプの変更、オブジェクトタイプの入力データソースの別のデータソースへの置き換えなどがあります。

ライブパイプラインが通常の周期で実行を続ける一方で、Funnel はユーザーに提供されるライブデータに影響を与えることなく、バックグラウンドで置き換え用パイプラインを統括します。置き換え用パイプラインの初回実行が成功すると、ライブパイプラインは破棄され、置き換え用パイプラインに置き換わります。オブジェクトタイプのスキーマとデータもそれに応じて更新されます。

Funnel が置き換え用パイプラインをプロビジョニングする最も一般的な理由はスキーマの変更ですが、パフォーマンス上の理由から、さまざまなヒューリスティクスに基づいて置き換え用パイプラインを自動的にプロビジョニングすることもあります。

インクリメンタル（差分処理）および完全な再インデックス化

以下のドキュメントは、正規の Object Storage v2 データストアに固有のものです。Object Storage v1（Phonograph）のインデックス化の動作については、OSv1 のドキュメントを参照してください。

インクリメンタルインデックス化（デフォルト）

正規の Object Storage v2 データストアは、データソースの新しいトランザクションごとにデータの差分を自動的に計算し、新しいデータ更新のみをインクリメンタルにインデックス化します。Funnel パイプラインは、デフォルトですべてのオブジェクトタイプにインクリメンタルインデックス化を使用します。インクリメンタルインデックス化により、すべてのデータを再度インデックス化する場合よりも Funnel パイプラインを迅速に実行できます。

たとえば、100行のデータソースに基づいている、100個のオブジェクトを持つオブジェクトタイプがあるとします。新しいデータ更新でそのうち10行が変更された場合、入力データソースのトランザクションタイプにかかわらず100個のオブジェクトすべてを再インデックス化するのではなく、Funnel バッチパイプラインは、変更された10行のみを含む新しい APPEND トランザクションを変更履歴データセットに作成します。

インクリメンタルデータセットのインクリメンタルインデックス化

Object Storage v2 は、インクリメンタルデータセットに基づいているオブジェクトタイプを同期する際に、「最新のトランザクションを優先する」方式を使用します。データセットに同じ主キーを持つ行が複数含まれている場合、最新のトランザクションの行のデータがオントロジーに反映されます。単一のトランザクション内に重複する主キーを含めることはできません。この動作は、ユーザー編集とデータソース更新の競合の処理方法とは関係ありません。

APPEND トランザクションを通じて行の更新を受け取るインクリメンタルデータセットを考えてみます。これは通常、変更履歴データセットと呼ばれます。同じデータの新しいバージョンは、値が更新され、主キーは同じである新しい行として表され、1回のトランザクションでデータセットに追加されます。is_deleted という名前の列は、デフォルトでは削除列として扱われません。削除列を使用するには、その列を識別するレガシー Object Storage v1 の変更履歴メタデータがデータソースのスキーマに含まれている必要があります。

Object Storage v2 は、変更履歴データセットを次のように同期します。

• 主キーが複数のトランザクションに出現する場合、最新のトランザクションの行が保持されます。
• 各トランザクションに含められる行は、主キーごとに最大1行です。
• 削除列は、レガシー Object Storage v1 の変更履歴メタデータで宣言されている場合にのみ考慮されます。

各トランザクションに含まれる行を主キーごとに最大1行にするには、変更履歴データセットにインクリメンタルウィンドウトランスフォームを実行する必要があると考えられます。

Copied!
1
2
3
4
5
6
from pyspark.sql.window import Window
from pyspark.sql import functions as F

ordering_window = Window().partitionBy('primary_key').orderBy(F.col('ordering').desc())
df = df.withColumn('rank', F.row_number().over(ordering_window))
df = df.filter((F.col('rank') == 1) & ~F.col('is_deleted'))

ビューのインクリメンタルインデックス化

データセットビューには、インクリメンタルインデックス化に関する制限があります。ビューは基盤となるデータ構造を抽象化するため、Funnel はコストを削減するためにインクリメンタルビルドの使用を試みます。ただし、次の制限が適用されます。

• 削除列を持つビューは、Funnel がどの行が削除されたかを判定できないため、常に完全に再インデックス化されます。
• 重複排除列の値がトランザクションの順序と競合する場合、インクリメンタルビルドで一貫性のない結果が生じる可能性があります。Funnel は「最後の編集を優先する」方式を使用します。重複排除列の値がより大きい行でも、インクリメンタルウィンドウの範囲外にある場合は無視されます。これを避けるには、削除列を追加して完全な再インデックス化が実行されるようにするか、トランザクションごとに重複排除列の値が常に増加するようにしてください。
• 基盤となるデータセットの最新のビューから削除する保持ポリシーは、古いデータが削除された場合に予期しない結果を引き起こす可能性があります。このような場合、完全な再インデックス化によって正しい結果が生成されます。

完全な再インデックス化（特殊なケース）

Funnel パイプラインは、次の場合にバッチインデックス化（すべてのオブジェクトを再インデックス化する処理）を使用します。

• 入力データソースの行の一定割合を超える行が同じトランザクションで変更された場合、変更履歴を計算してインクリメンタルにインデックス化するよりも、再インデックス化の方が計算コストを抑えられ、より高速になることがあります。デフォルトのしきい値は、同じトランザクションで行の80%が変更された場合に設定されています。
• オブジェクトタイプのスキーマに対する特定の変更により、Funnel 置換パイプラインが必要になる場合。このパイプラインは、バックグラウンドでまったく新しい Funnel パイプライン（OSv2 インデックスを含む）を作成します。
• ユーザーが Ontology Manager から完全な再インデックス化を実行する場合。

Funnel パイプラインを監視する

Funnel パイプラインは複数のビルドジョブで構成されています。監視ビューでは、監視ルールを作成することで、Funnel パイプライン内の特定のジョブのヘルスを追跡できます。

ユーザーは、Ontology Manager でこのオブジェクトタイプのヘルスを監視を選択することで、監視ビューを作成できます。これにより、以下のスクリーンショットに示すように、Data Health アプリケーションの監視ビュータブに移動します。

監視ビュータブから、ライブパイプラインと置換パイプラインの両方のジョブを監視するルールを作成できます。また、同期伝播遅延ルールを追加すると、オブジェクトデータベース内のインデックス化済みデータの鮮度がルールで定義されたしきい値を超えたときに通知を受け取れます。

一方、Object Storage v1（Phonograph）は、ヘルスチェックを使用してオントロジーリソースの同期を監視します。OSv1 では、オブジェクトタイプの同期ジョブは1個だけであり、ユーザーは同期ジョブに直接これらのヘルスチェックを定義できます。

パイプラインをデバッグする

Foundry のビルドジョブは、さまざまな理由で失敗することがあります。オブジェクトタイプの元データソースに対する閲覧権限を持つユーザーは、オブジェクトタイプのデータソースタブにあるライブパイプラインダッシュボードで、パイプラインのエラーを確認できます。パイプライングラフで失敗したジョブを選択し、以下のスクリーンショットに示すように失敗したジョブを選択します。

または、Builds アプリケーションに移動し、左側のパネルの検索フィルターでオブジェクトタイプによる絞り込みを行うことで、特定のオブジェクトタイプのすべてのビルドジョブを一覧表示できます。