データパイプラインの主要概念 ​

Workatoデータパイプラインは、正確で最新のデータセットを維持するために、データを抽出、複製、同期します。パイプラインは、ソースアプリケーションを宛先データウェアハウスに接続し、データを一括で移動して、スキーマの整合性を維持します。次のセクションでは、データパイプラインがデータを処理および管理する方法を説明する主要概念を定義します。

ソースアプリケーションと宛先 ​

データパイプラインは、Salesforceなどのソースアプリケーションからデータを抽出し、Snowflakeなどの宛先に同期します。 1つのパイプラインは、ソースアプリケーション内の複数のオブジェクトまたはフィールドからデータを取得し、そのデータを指定された宛先にレプリケートします。

接続性 ​

Workatoは、Stripe、Slack、QuickBooks Onlineなどのクラウドソースに対して、オンプレミスエージェントを使用せずにHTTPS経由で接続します。オンプレミスデータベースや、インターネットに公開されていないSAPシステムなど、プライベートネットワーク内で動作するソースは、Workatoのオンプレミスエージェント経由で接続します。どの接続方法をサポートしているかを確認するには、ソースアプリケーションの設定ガイドを参照してください。

オブジェクトラベルとAPI名 ​

Microsoft Dynamics 365、Microsoft Dynamics Business Central、ServiceNowなどのソースでは、各オブジェクトに対して、人間が読みやすいラベルとAPI名または論理名の両方が表示されます。例えば、Microsoft Dynamics 365ではAccount (account)のように表示されます。ラベルはソースアプリケーションで表示される名前であり、API名または論理名はソースのAPIが使用する識別子です。

名前変更時の動作はソースによって異なります。例えば、Workatoは、Google Sheetsのタブ名を変更した後も引き続き同期を行いますが、Airtableのフィールド名が変更されると、そのテーブルのスキーマを次回検出したときに別のフィールドとして扱います。どの動作が適用されるかを確認するには、ソースアプリケーションの設定ガイドを参照してください。

オブジェクト同期 ​

同期とは、パイプラインがソースからデータを抽出し、宛先にロードする全体的なプロセスを指します。各同期は複数のオブジェクトを並列に処理し、次のいずれかのタイプを使用します：

• Full sync

• ソースから利用可能なすべてのレコードを抽出し、宛先にロードします。これにより、宛先テーブルには同期時点のソースデータの完全なスナップショットが含まれます。

• Incremental sync

• オブジェクトのインクリメンタルカーソルに基づき、最後に同期が成功してから新規作成または更新されたレコードを抽出します。削除されたレコードは、削除トラッキングをサポートするオブジェクトについてのみ同期されます。

• Re-sync

• 特定のオブジェクトに対する手動の1回限りの同期。 Re-syncは、そのオブジェクトの現在のデータをただちに抽出してロードします。このタイプは、エラー、スキップされた同期、またはデータの変更により、1つのオブジェクトのみを再度同期する必要がある場合に便利です。

データパイプラインは、完全な履歴同期から開始されます。インクリメンタル同期を使用するオブジェクトは、初回の同期が完了するとインクリメンタル同期に切り替わります。フル同期を使用するオブジェクトは、オブジェクトの同期モードをFull syncに設定した場合、またはオブジェクトがインクリメンタル同期をサポートしていない場合、スケジュールされた各実行でフル同期を実行します。 Re-syncを使用すると、通常のスケジュール外で特定のオブジェクトを再処理できます。

各オブジェクトで利用できる同期モードのオプションはソースによって異なります。 Salesforce、Braintree、Freshdeskなどのソースでは、オブジェクトをパイプラインに追加する際に、ドロップダウンのSync modeでFull syncまたはIncrementalを選択して同期モードを指定できます。これらのソースは、タイムスタンプなどの変更を追跡する手段が提供されていないオブジェクトについては、そのオブジェクトのデフォルトをFull syncに設定します。 SAP Concurは、オブジェクトに対して使用可能な更新時刻フィルターをAPIが公開しているかどうかに基づき、自動的に同期モードを設定します。 Google Sheetsはフル同期のみをサポートします。

詳細については、Syncタイプと実行ガイドを参照してください。

主キーとインクリメンタルカーソル ​

APIおよびデータベースソースのインクリメンタル同期は、一般的に各オブジェクトの2つのプロパティに依存します。

• インクリメンタルカーソル: パイプラインが、最後に同期が成功してから新規作成または更新されたレコードを見つけるために使用するフィールドであり、通常は最終更新タイムスタンプです。オブジェクトによっては、PagerDutyのcreated_atのような作成タイムスタンプや、Gongのuser_activity_statsオブジェクトのような日付範囲を使用する場合もあります。
• 主キー: レコードIDなどの一意の識別子であり、宛先側で既存行を更新して重複行の挿入を避けるために使用されます。

ChargebeeのCouponCodeやGiftオブジェクトのように、利用可能なカーソルまたは主キーが存在しない場合、そのオブジェクトはフル同期のみに制限されることがあります。 BraintreeやApollo.ioなどのソースは、各インクリメンタル実行で最近の時間ウィンドウを再読み込みします。例えば、Braintreeは、作成タイムスタンプのカーソルでは反映されないステータス変更を捕捉するために、直近3日間のウィンドウを再読み込みします。

ファイルベースのソースは、個々のレコードではなくファイルを追跡します。例えば、SFTPとOneDriveは、最後に記録された更新時刻以降に作成または更新されたファイルを処理します。

各オブジェクトでサポートされる同期モードについては、ソースアプリケーションの設定ガイドを参照してください。

履歴開始日 ​

When first started, this pipeline should pick up records fromフィールドは、履歴同期中にパイプラインがレコードを抽出する最も古い日付を設定します。このフィールドを空のままにすると、パイプラインは利用可能なすべてのレコードを抽出します。パイプラインの実行後にこの値を変更することはできません。

この日付がフル同期オブジェクトに与える影響はソースによって異なります。 OutreachやSage Intacctなどのソースでは、この日付はインクリメンタル同期をサポートするオブジェクトにのみ適用されるため、フル同期オブジェクトは常にすべてのレコードを抽出します。 Apollo.ioは、callsなどのオブジェクトについて、この日付を使用してフル同期の実行範囲も制限します。動作の詳細については、ソースアプリケーションの設定ガイドを参照してください。

ソースフィルター ​

Salesforce、Snowflake、Slackなどのソースでは、同期対象とするレコードを絞り込むことができます。 Salesforceオブジェクトの設定ではSOQL WHERE clauseフィールドにSOQL条件を入力し、Snowflake as a data pipeline sourceを設定する際にはWHERE clauseフィールドにSQL条件を入力し、あるいはSlackパイプラインを特定のチャンネルリストに限定できます。

ソースフィルターは、すべてのソースやオブジェクトで利用できるわけではありません。ソースアプリケーションがソースフィルターをサポートしているかどうかを確認するには、その設定ガイドを参照してください。

同期頻度 ​

データパイプラインはスケジュール設定された頻度で実行され、ソースシステムと宛先システムの同期を維持します。デフォルトの頻度は15分です。 5分間隔はリクエストに応じて利用でき、顧客ごとに有効にできます。

同期は、時間ベースの間隔またはcron式でスケジュールできます。スケジュールオプションはソースによって異なります。スケジュールされた各実行では、選択されたすべてのオブジェクトが同期され、インクリメンタル同期を使用するオブジェクトについてはインクリメンタル同期が実行されます。運用要件とシステム容量に合ったスケジュールを選択します。

現在の同期が想定より長く実行され、次回のスケジュールと重なった場合、パイプラインは次のスケジュールされた同期をスキップします。

パイプライン実行 ​

各データパイプライン同期は複数の実行で構成され、選択したオブジェクトごとに1つの実行があります。同期はすべてのオブジェクトに対するアクティビティ全体を表し、実行はその同期内の単一オブジェクトの実行状況を追跡します。

パイプラインはパフォーマンスを向上させるために実行を並列で実行します。各実行はソースからデータを抽出し、宛先にロードします。実行レベルのデータはRunsタブに表示され、パイプラインの実行を監視するのに役立ちます。詳細については、オブジェクト実行セクションを参照してください。

同時実行制限 ​

パイプラインを設定する際に、オプションのConcurrency limitフィールドを使用して、パイプラインがソースに対して実行する同時実行処理数の上限を指定します。これは、同時リクエスト数に制限があるソースで役立ちます。たとえば、一度にアクティブセッションを1つしか受け付けないSFTPサーバーなどです。このフィールドを空のままにすると、パイプラインはWorkatoで設定されたデフォルトのクォータを使用します。

スキーマレプリケーションとスキーマドリフト管理 ​

スキーマドリフトとは、ソースデータに変更が発生したときに生じる、ソースと宛先の間の不整合を指します。これらの変更には、フィールドの追加や削除、フィールドタイプの変更、またはその他の構造的な更新が含まれる場合があります。管理されていないスキーマドリフトは、変換エラー、データ損失、不正確な分析の原因となる可能性があります。

Workatoパイプラインは同期中にスキーマドリフトを検出し、パイプライン設定に基づいてスキーマ変更を適用します。 Auto-sync new fieldsオプションを使用すると、新しいソースフィールドを自動的に同期できます。一方、Block new fieldsを使用すると、スキーマを固定し、新しいフィールドを手動で追加できます。この動作は、パイプラインの設定時に構成できます。

パイプラインは、インクリメンタル同期中に新しいフィールドの履歴値をバックフィルしません。フィールドを検出した時点以降の値のみを取得します。

次のスキーマ変更については、宛先側で手動更新が必要です。

• 削除されたフィールド: ソースでフィールドが削除された場合でも、パイプラインは履歴参照のために宛先側のテーブルにそのフィールドの列を保持します。フィールドとそのデータを削除するには、宛先テーブルからその列を手動で削除してください。 Deleted fields in the sourceを参照してください。
• Salesforceでのデータ型変更: Salesforceでフィールドのデータ型が変更されても、パイプラインは宛先スキーマを更新しないため、宛先スキーマは手動で更新する必要があります。 Schema handlingを参照してください。

削除追跡 ​

削除トラッキングは、ソースに存在しなくなったレコードをマークして、挿入および更新だけでなく削除も宛先に反映させます。 Workatoは通常、宛先テーブルで削除されたレコードに対して、合成列_workato_is_deletedをtrueに設定します。

Workatoはソースに応じて、次のいずれかの方法で削除を検出します。

• ネイティブな削除シグナル: Chargebeeレコードのdeletedフラグのように、レコードが削除されたことを示すフラグをソースが公開している場合です。
• 宛先側での推測: ソースが削除シグナルを公開していない場合です。 Workatoは各フル同期の結果を前回の実行と比較し、もはや存在しないレコードをマークします。
• 変更トラッキング: SQL Server Change TrackingやServiceNowのsys_audit_deleteテーブルのように、ソースが提供する変更トラッキング機構をWorkatoが読み取り、削除を検出する場合です。

削除トラッキングは、すべてのソースやオブジェクトで利用できるわけではありません。例えば、SFTPは削除トラッキングをサポートしていません。どのような削除トラッキング動作をサポートしているかを確認するには、ソースアプリケーションの設定ガイドを参照してください。

宛先スキーマとテーブル構成 ​

Snowflake、Databricks、PostgreSQLを宛先とする場合、パイプラインがテーブルを作成するスキーマを選択できます。 BigQueryではデータセットを使用し、S3 Icebergでは名前空間を使用します。 SnowflakeとPostgreSQLでは、コネクションで設定されたスキーマを上書きするパイプライン専用のスキーマも設定できます。 PostgreSQLは既存のスキーマのみをサポートします。

Snowflakeでは、すべてのテーブル名の先頭に値を追加するTable prefixも設定できます。テーブルプレフィックスを使用すると、複数のパイプラインが同じスキーマに書き込む場合でも、テーブル名の競合を防ぐことができます。 Snowflakeでは、パイプラインが一度実行された後は、SchemaやTable prefixを変更できません。

パイプラインを開始する前に、宛先スキーマ、BigQueryデータセット、S3 Icebergのウェアハウスおよびカタログが新規作成されており、空であることを確認してください。これにより、初回同期中のエラーを防ぎ、パイプラインが競合なしで送信先テーブルを作成できるようになります。宛先でサポートされるオプションについては、その設定ガイドを参照してください。

宛先テーブルとマージ戦略 ​

パイプラインは、ソーススキーマに基づいて宛先テーブルを作成します。 Snowflake、Databricks、BigQuery、PostgreSQL、SQL Serverを宛先とする場合、パイプラインはデータレプリケーションと更新処理を行うためのステージおよび一時テーブルも作成するため、宛先コネクションにはそれらを作成する権限が必要です。

Amazon S3やSFTPなどのファイルベースソースでは、Choose a merge strategyフィールドで、行を宛先テーブルにどのようにマージするかを選択できます。

• アップサート: 新しい行を挿入し、既存の行を更新します。 Merge methodフィールドを使用して、主キーとして使用する列を1つ以上選択します。このフィールドを空のままにすると、パイプラインはシステム生成列_fileおよび_lineを複合主キーとして使用します。
• Append only: すべての行を挿入し、既存レコードの突き合わせや更新は行いません。

合成列 ​

Workatoは、削除されたレコードのマーク、パイプライン実行の追跡、ソースファイルの特定を行うために、宛先テーブルに合成列を追加します。追加される列はソースによって異なりますが、次の列が含まれます。

ソースアプリケーションが追加する合成列については、その設定ガイドを参照してください。

フィールドレベルのデータ保護 ​

フィールドレベルのデータ保護は、同期中の機密情報を保護するのに役立ちます。各フィールドについて、次のいずれかのオプションを選択できます。

• そのまま複製: ソースのデータ値が宛先に同一に複製されます。
• Hash: Workatoがフィールドの機密データ値を宛先に同期する前にハッシュします。このオプションでは、データ構造を維持しつつ、内容を保護します。

Workatoは、値を宛先にロードする前に、転送中の各値をハッシュするために次の手順を実行します。

• 宛先固有の一意のソルトをフィールド値に付加し、NULL値は空文字列として扱います。
• 結果をUTF-8としてエンコードし、SHA-256アルゴリズムでハッシュします。
• ハッシュをBase64エンコードされた文字列として宛先に書き込みます。

同じ値は同一の宛先内では常に同じハッシュを生成するため、その宛先内のテーブル間でハッシュ化された列をジョインできます。同じ値であっても、各宛先には固有のソルトがあるため、宛先ごとに異なるハッシュが生成されます。ソルトがあっても、ハッシュ化された値から元の値を復元することはできません。

パイプラインがフィールドを同期した後に、そのフィールドをReplicate as isとHashの間で切り替えると、Workatoは宛先列の既存値をクリアします。フィールドをHashに切り替えると、Workatoはハッシュ値に適したデータ型を持つ列に置き換えます。その後の同期から、選択した形式で値がロードされます。ソース側でフィールドのデータ型が後から変更されても、Workatoはそのフィールドの値を引き続きハッシュします。

Workatoは、個人を特定できる情報（PII）やその他の機密フィールドをハッシュすることを推奨します。 pipeline setup中にオブジェクトを追加する際、各フィールドごとにフィールドレベルのデータ保護を設定できます。 Workatoは、フィールドレベルのデータ保護に関する変更を監査ログに記録します。