更新ポリシーの概要

  • [アーティクル]

更新ポリシーは、新しいデータがテーブルに書き込まれるときにトリガーされる自動化メカニズムです。 取り込まれたデータを変換し、結果を変換先テーブルに保存するクエリを実行することで、特別なオーケストレーションが不要になります。 1 つのテーブルに複数の更新ポリシーを定義できるため、異なる変換を行い、同時に複数のテーブルにデータを保存できます。 ターゲット テーブルには、ソース テーブルとは異なるスキーマ、アイテム保持ポリシー、およびその他のポリシーを含めることができます。

たとえば、高速トレース ソース テーブルには、フリーテキスト列として形式設定されたデータを含めることができます。 ターゲット テーブルには、parse 演算子を使用してソース テーブルのフリーテキスト データの変換から生成され、適切に構造化されたスキーマに従って、特定のトレース行を含めることができます。 詳細については、 一般的なシナリオを参照してください。

次の図は、更新ポリシーの概要を示しています。 2 つ目のソース テーブルに データが追加されたときにトリガーされる 2 つの更新ポリシーが表示され、変換されたデータが 2 つのターゲット テーブルに追加されます。

図は、更新ポリシーの概要を示しています。

更新ポリシーには、通常のインジェストと同じ制限とベスト プラクティスが適用されます。 ポリシーはクラスターのサイズに応じてスケールアウトされ、一括インジェストを処理する場合に効率が向上します。

Note

  • ソースとターゲット テーブルは、同じデータベース内にある必要があります。
  • 更新ポリシー関数スキーマとターゲット テーブル スキーマは、列名、型、順序で一致している必要があります。

形式設定されたデータを取り込む際は、パフォーマンスが向上します。CSV が推奨されます。これは形式が適切に定義されているためです。 ただし、データの形式を制御できない場合や、データベース内の静的ディメンション テーブルとレコードを結合するなどして、取り込まれたデータを強化したい場合があります。

更新ポリシーのクエリ

更新ポリシーがターゲット テーブルで定義されている場合、ソース テーブルに取り込まれているデータに対して複数のクエリを実行できます。 複数の更新ポリシーがある場合、実行順序が必ずしもわかっているわけではありません。

クエリの制限事項

  • ポリシー関連のクエリでは、ストアド関数を呼び出すことができますが、次のことができます。
    • クラスター間クエリを実行することはできません。
    • 外部データや外部テーブルにはアクセスできません。
    • (プラグインを使用して) 吹き出しを作成することはできません。
  • クエリには、 RestrictedViewAccess ポリシー が有効になっているテーブルへの読み取りアクセス権がありません。
  • ストリーミング インジェストにおける更新ポリシーの制限については、ストリーミング インジェストの制限事項に関するセクションを参照してください。

警告

クエリが正しくないと、ソース テーブルへのデータ インジェストが妨げられる可能性があります。 クエリ結果とソース テーブルと変換先テーブルのスキーマの互換性だけでなく、制限によって、ソース テーブルへのデータ インジェストを防ぐクエリが正しくない可能性があることに注意することが重要です。

これらの制限は、ポリシーの作成と実行中に検証されますが、クエリが参照する可能性がある任意のストアド関数が更新される場合は検証されません。 そのため、更新ポリシーをそのまま維持するには、注意して変更を加える必要があります。

ポリシーの Query 部分、または Query 部分によって参照されている関数内で Source テーブルを参照する場合:

  • テーブルの修飾名を使用しないでください。 代わりに TableName を使用してください。
  • database("DatabaseName").TableName も cluster("ClusterName").database("DatabaseName").TableName も使用しないでください。

更新ポリシー オブジェクト

テーブルには、0 個以上の更新ポリシー オブジェクトを関連付けることができます。 そうした各オブジェクトは、次のプロパティが定義された JSON プロパティ バッグとして表されます。

プロパティ Type 説明
IsEnabled bool 更新ポリシーが有効 (true) か、無効 (false) かを示す状態
source string 更新ポリシーの呼び出しをトリガーするテーブルの名前
クエリ string 更新用のデータを生成するために使用されるクエリ
IsTransactional bool 更新ポリシーがトランザクションであるかどうかを示します。既定値は false です。 ポリシーがトランザクションであり、更新ポリシーが失敗した場合、ソース テーブルは更新されません。
PropagateIngestionProperties bool エクステント タグや作成時間など、ソース テーブルへのインジェスト中に指定されたプロパティがターゲット テーブルに適用される場合の状態。
ManagedIdentity string 更新ポリシーを実行する代わりにマネージド ID。 マネージド ID には、オブジェクト ID または予約語を system 指定できます。 クエリが他のデータベース内のテーブルを参照する場合、または 行レベルのセキュリティ ポリシーが有効になっているテーブルを参照する場合は、更新ポリシーをマネージド ID で構成する必要があります。 詳細については、「 マネージド ID を使用して更新ポリシーを実行する」を参照してください。

Note

実稼働システムでは、一時的な障害により、ターゲット テーブルでデータが失われないようにするために、IsTransactionaltrue を設定します。

Note

テーブル A からテーブル B へ、テーブル C へなどのカスケード更新は許可されます。ただし、更新ポリシーが循環的な方法で定義されている場合は、実行時にこれが検出され、更新のチェーンが切断されます。 データは、チェーン内の各テーブルに 1 回だけ取り込まれます。

管理コマンド

ポリシー管理コマンドの更新には、次のものが含まれます。

更新ポリシーはインジェスト後に開始される

更新ポリシーは、次の任意のコマンドを使用して、データがソース テーブルに取り込まれたり移動したり、ソース テーブルにエクステントが作成されたりした場合に有効になります。

警告

.set-or-replace コマンドの一環として更新ポリシーが呼び出された場合、既定では、派生テーブルのデータはソース テーブルと同じ方法で置き換えられます。 replace コマンドが呼び出された場合、更新ポリシーのリレーションシップが設定されているすべてのテーブルでデータが失われる可能性があります。 代わりに .set-or-append を使用することを検討してください。

ソース テーブルからデータを削除する

ターゲット テーブルにデータを取り込んでから、必要に応じてソース テーブルからデータを削除できます。 ソース テーブルのアイテム保持ポリシーで論理的な削除期間を 0sec (または 00:00:00) に、さらに更新ポリシーをトランザクションとして設定します。 次の条件が適用されます：

  • ソース テーブルからソース データに対してクエリを実行することはできません
  • インジェスト操作の一環としてソース データが永続ストレージに保持されることはありません
  • 操作パフォーマンスが向上します。 ソース テーブルのエクステントに対するバックグラウンド グルーミング操作のために、インジェスト後のリソースが削減されます。

注意

ソース テーブルの論理的な削除期間が ( または 00:00:00) の0sec場合、このテーブルを参照するすべての更新ポリシーはトランザクションである必要があります。

パフォーマンスへの影響

更新ポリシーはクラスターのパフォーマンスに影響を与える可能性があります。また、データ エクステントのインジェストはターゲット テーブルの数に応じて増えます。 ポリシー関連のクエリを最適化することが重要です。 更新ポリシーのパフォーマンスへの影響をテストするには、ポリシーを作成または変更する前に既に存在するエクステントに対して、またはクエリで使用される関数に対してポリシーを呼び出します。

リソースの使用状況を評価する

.show queries と次のパラメーターを使用して、リソースの使用状況 (CPU、メモリなど) を評価します。

  • Source プロパティ (ソース テーブル名) を MySourceTable として設定します
  • MyFunction() という名前の関数を呼び出すように Query プロパティを設定します
// '_extentId' is the ID of a recently created extent, that likely hasn't been merged yet.
let _extentId = toscalar(
    MySourceTable
    | project ExtentId = extent_id(), IngestionTime = ingestion_time()
    | where IngestionTime > ago(10m)
    | top 1 by IngestionTime desc
    | project ExtentId
);
// This scopes the source table to the single recent extent.
let MySourceTable =
    MySourceTable
    | where ingestion_time() > ago(10m) and extent_id() == _extentId;
// This invokes the function in the update policy (that internally references `MySourceTable`).
MyFunction

エラー

の既定の IsTransactional:false設定では、ポリシーが実行されない場合でも、ソース テーブルにデータを取り込むことができます。

を設定 IsTransactional:true すると、ソース テーブルとターゲット テーブル内のデータの一貫性が保証されます。 ただし、ポリシー条件が失敗した場合、データはソース テーブルに取り込まれません。 また、条件によっては、データがソース テーブルには取り込まれるが、ターゲット テーブルには取り込まれない場合があります。 ただし、ポリシーが正しく定義されていない場合、またはスキーマが一致しない場合、データはソース テーブルまたはターゲット テーブルに取り込まれません。 たとえば、クエリ出力スキーマとターゲット テーブルの不一致は、ターゲット テーブルから列を削除した場合に発生する可能性があります。

.show ingestion failures コマンドを使用して、失敗を確認できます。

.show ingestion failures
| where FailedOn > ago(1hr) and OriginatesFromUpdatePolicy == true

失敗の処理

非トランザクション ポリシー

に IsTransactional:false設定すると、ポリシーの実行に失敗した場合は無視されます。 インジェストは自動的には再試行されません。 手動でインジェストを再試行できます。

トランザクション ポリシー

にIsTransactional:true設定すると、インジェスト メソッドが の場合、pullデータ管理 サービスが関係し、次の条件に従ってインジェストが自動的に再試行されます。

  • 再試行は、次のいずれかの構成可能な制限設定が満たされるまで実行されます。DataImporterMaximumRetryPeriod または DataImporterMaximumRetryAttempts
  • 既定では、設定 DataImporterMaximumRetryPeriod は 2 日間で DataImporterMaximumRetryAttempts、10 です
  • バックオフ期間は 2 分から始まり、2 倍になります。 そのため、待機は 2 分から始まります。その後、4 分、8 分、16 分というように増加します。

それ以外の場合は、インジェストを手動で再試行できます。

抽出、変換、読み込みの例

更新ポリシー設定を使用して、抽出、変換、読み込み (ETL) を実行できます。

この例では、単純な関数で更新ポリシーを使用して ETL を実行します。 まず、2 つのテーブルを作成します。

  • ソース テーブル - データを取り込む 1 つの文字列型の列が含まれています。
  • ターゲット テーブル - 目的のスキーマが含まれています。 このテーブルには更新ポリシーが定義されています。

  1. ソース テーブルを作成しましょう。

    .create table MySourceTable (OriginalRecord:string)

  2. 次に、ターゲット テーブルを作成します。

    .create table MyTargetTable (Timestamp:datetime, ThreadId:int, ProcessId:int, TimeSinceStartup:timespan, Message:string)

  3. その後、データを抽出する関数を作成します。

    .create function
 with (docstring = 'Parses raw records into strongly-typed columns', folder = 'UpdatePolicyFunctions')
     ExtractMyLogs()
    {
    MySourceTable
    | parse OriginalRecord with "[" Timestamp:datetime "] [ThreadId:" ThreadId:int "] [ProcessId:" ProcessId:int "] TimeSinceStartup: " TimeSinceStartup:timespan " Message: " Message:string
    | project-away OriginalRecord
}

  4. ここで、作成した関数を呼び出すように更新ポリシーを設定します。

    .alter table MyTargetTable policy update
@'[{ "IsEnabled": true, "Source": "MySourceTable", "Query": "ExtractMyLogs()", "IsTransactional": true, "PropagateIngestionProperties": false}]'

  5. データがターゲット テーブルに取り込まれた後にソース テーブルを空にするために、ソース テーブルにアイテム保持ポリシーを定義して、その SoftDeletePeriod として 0s を設定します。

     .alter-merge table MySourceTable policy retention softdelete = 0s

関連コンテンツ