カスタム JDBC ソース

Foundry を JDBC ソースに接続すると、Foundry とほとんどのリレーショナルデータベースやデータウェアハウスとの間でデータを読み取り、同期できます。

ソースを設定すると、抽出定義に指定した SQL クエリを使用して、データベースからデータを同期する方法を柔軟に定義できます。また、前回の同期実行後に更新されたデータのみを読み取るインクリメンタル（差分）同期を設定できます。

セットアップ

1. Data Connection アプリケーションを開き、画面の右上にある + 新規ソースを選択します。
2. 一覧に表示された種類から、該当するソースを探します（たとえば、BigQuery や PostgreSQL）。コネクターの種類が一覧にない場合は、代わりに JDBC を選択します。
3. 追加の設定プロンプトに従い、以下のセクションの情報を使用してコネクターのセットアップを続けます。

設定オプション

JDBC ドライバー

デフォルトでは、Data Connection の Agent は JDBC に対応するシステムに接続する機能を備えています。ただし、システムに正常に接続するには、JDBC ドライバー ↗を指定する必要があります。また、ドライバークラスを指定する必要があります。通常、ドライバークラスは該当するドライバーの公開ドキュメントに記載されています。たとえば、Snowflake の JDBC ドライバーのクラスは、公開ドキュメント ↗で確認できます。

Foundry ワーカー接続

Foundry ワーカー接続では、Palantir が署名したドライバーは必要ありません。通常、Google BigQuery ↗など、該当するソースの種類のドキュメントからドライバーを見つけてダウンロードできます。その後、接続を設定するときにドライバーをアップロードできます。

Agent ワーカー接続 [レガシー]

Agent ワーカーソースでは、ドライバーを Agent に手動でアップロードする必要があります。ただし、Agent にアップロードした JDBC ドライバーは、Agent にすでに存在するほかの JDBC ドライバーとクラスパスの競合を引き起こす場合があります。競合の可能性を避けるため、JDBC ドライバーを手動でアップロードする代わりに、Palantir が提供するドライバーを使用します。

Agent にドライバーを割り当てるには、以下の手順に従います。

1. Data Connection で Agent の概要ページに移動し、Agent 設定タブを選択します。
2. JDBC ドライバーセクションに移動し、編集を選択します。

3. 必要なドライバーが右側のサイドバーにない場合は、右上の新しいものをアップロードを選択して、新しいドライバー JAR ファイルをアップロードします。ドライバーを保存する場所を選択します。
4. ドライバー名の横にある ⊕ ボタンを選択して、Agent にドライバーを追加します。

5. ⊕ を選択して、Agent にドライバーを追加し、サイドバーを閉じます。

6. 必要なドライバーを選択またはアップロードした後、新しいドライバーファイルを使用するには、Agentを再起動を選択する必要があります。

JDBC プロパティ

必要に応じて、JDBC 接続にプロパティ ↗を追加して、動作を設定できます。接続設定に追加できるその他の JDBC プロパティについては、該当するソースのドキュメントを参照してください。

出力のオーバーライド：ソース

必要に応じて出力のオーバーライドを追加し、出力ファイルの形式と JDBC 同期方式を変更できます。設定した内容は、そのソースを使用するすべての JDBC 同期に適用されます。同期設定を編集すると、特定の同期に対してこれらのパラメーターをオーバーライドできます。出力のオーバーライドを追加するには、セットアップページの下部にある高度なオプションセクションを展開し、以下を入力します。

フェッチサイズ

出力のフェッチサイズは、クエリでデータベースとの1回の往復ごとに取得する行数です。フェッチサイズを調整すると、同期ごとのネットワーク呼び出しの合計回数を変更できます。ただし、フェッチサイズはメモリー使用量に影響します。フェッチサイズを大きくすると、メモリー使用量が増える代わりに、同期が高速になります。フェッチサイズを500から始め、状況に応じて調整することを推奨します。

最大ファイルサイズ

出力ファイルの最大ファイルサイズ（バイト数または行数）も調整できます。これにより、Foundry へのデータアップロードのパフォーマンスと耐障害性が向上する場合があります。

ファイルサイズを Bytes で指定する場合、バイト数は Parquet ライター（128MB）または Avro ライター（64KB）のメモリー内バッファーサイズの少なくとも2倍である必要があります。

JDBC ソースからデータを同期する

JDBC 同期を設定するには、ソースの概要画面の右上にあるソースを探索して同期を作成を選択します。次に、Foundry に同期するテーブルを選択します。同期の準備ができたら、x個のデータセットの同期を作成を選択します。

Foundry のソース探索については、こちらを参照してください。

JDBC 同期を設定する

事前クエリ

事前クエリは、実際の SQL クエリが実行される前に実行される SQL クエリの配列で、任意で指定できます。実際のクエリを実行する前にデータベースのリフレッシュをトリガーする必要があるユースケースでは、事前クエリの使用を推奨します。

SQL クエリ

同期ごとに1つの SQL クエリを実行できます。このクエリは、データのテーブルを出力として生成する必要があり、ストアドプロシージャの呼び出しなどの操作を実行してはいけません。クエリの結果は、Foundry の出力データセットに保存されます。

出力のオーバーライド：同期

ソース設定レベルで出力のオーバーライドを設定するほかに、個々の同期出力に特定のオーバーライドを適用することもできます。保存した設定は、その同期にのみ適用されます。設定オプションの詳細については、上記の出力のオーバーライドセクションを確認してください。

精度の制限

同期設定レベルでは、個々の JDBC 同期に対して精度の制限を適用することを選択できます。この制限により、精度が38桁を超える数値は拒否されます。この設定はデフォルトで無効になっています。

JDBC 同期を最適化する

新しい同期を設定する場合や、パフォーマンスの問題に対処する場合は、同期の速度と信頼性を向上させるために、インクリメンタル（差分）同期への切り替えや SQL クエリの並列化を検討してください。

まず、インクリメンタル同期を試すことを推奨します。問題が解消されない場合は、SQL クエリの並列化に進みます。

インクリメンタル JDBC 同期

通常、同期では、同期の間にデータが変更されたかどうかにかかわらず、対象テーブルから一致するすべての行がインポートされます。これに対し、インクリメンタル同期は状態を保持する同期であり、データを重複させずに同じテーブルから APPEND 形式のトランザクションを実行できます。

インクリメンタル同期は、JDBC ソースから大きなテーブルを取り込む場合に使用できます。インクリメンタル同期を使用するには、テーブルに厳密に単調増加する列が含まれている必要があります。

インクリメンタル JDBC 同期を設定するには、以下の手順に従います。

1. 同期を編集ページで、トランザクションの種類を APPEND に設定します。
2. 次に、インクリメンタルセクションで有効にするを選択します。
3. 続いて、単調増加する列と、その列の初期値を指定します。インポートする予定のどの値よりも小さい値を選択します。インクリメンタル同期では、すでにインポートされた最大値よりも値が大きい行がインポートされます。

例： JDBC ソースに同期する5 TB のテーブルに、数十億行が含まれています。このテーブルには、id という単調増加する列があります。id 列をインクリメンタル列として使用し、初期値を -1、行数の上限を5000万行に設定することで、1回に5000万行ずつ取り込むように同期を設定できます。

同期が最初に実行されると、id の値が -1 より大きい最初の5000万行（id に基づく昇順）が Foundry に取り込まれます。たとえば、この同期を数回実行し、前回の同期で取り込まれた最大の id 値が19384004822だった場合、次の同期では、id 値が19384004822より大きい最初の行から、次の5000万行が取り込まれます。以降も同様です。

SQL クエリにも必ず上限を追加してください。たとえば、クエリが SELECT * FROM "sales"."region" の場合、SELECT * FROM "sales"."region" WHERE sale_id > ? limit 100000 に変更できます。ビルドが実行されるたびに、100000行が Foundry にインポートされます。クエリの ? の値は、前回の実行時の値で自動的に更新されます。

SQL クエリを並列化する

インクリメンタル同期に切り替えてもパフォーマンスが改善しない場合は、SQL クエリを並列化して、小さなクエリに分割できます。分割されたクエリは、Agent によって並列に実行されます。

これを実現するには、SQL クエリを新しい構造に変更する必要があります。以下に例を示します。

SELECT
/* FORCED_PARALLELISM_COLUMN(<column>), FORCED_PARALLELISM_SIZE(<size>) */
*
FROM <table_name>

クエリの並列化に必要な詳細を以下で説明します。

FORCED_PARALLELISM_COLUMN(<column>)：テーブルの分割に使用する列を指定します。できるだけ均等に値が分布している数値列（または数値列を生成する列の式）を指定する必要があります。

FORCED_PARALLELISM_SIZE(<size>)：並列度を指定します。たとえば、4 を指定すると、5つのクエリが同時に実行されます。4つのクエリは、並列化用に指定した列の値を分割して処理し、もう1つのクエリは、その列の NULL 値を検索します。

たとえば、上記の SQL クエリ SELECT * FROM sales_data は、追加の情報を含めることで並列化できます。

SELECT
/* FORCED_PARALLELISM_COLUMN(sales_id), FORCED_PARALLELISM_SIZE(4) */
*
FROM "sales"."region"

Iceberg 同期

カスタム JDBC ソースから Foundry の Iceberg テーブルに直接データを同期できます。Iceberg 同期は、非インクリメンタルとインクリメンタルの両方のバッチ取り込み、および複数の書き込みモードをサポートし、Foundry ワーカーのランタイムで実行されます。設定の詳細とサポートされる書き込みモードについては、Iceberg 同期のドキュメントを確認してください。

JDBC ソースにデータをエクスポートする

Data Connection は、スキーマを持つデータセットを JDBC ソースにテーブルエクスポートする機能をサポートしています。

JDBC エクスポートを有効にし、設定し、スケジュールする方法については、ドキュメントを確認してください。

JDBC ソースのエクスポート設定オプション

JDBC ソースを使用したテーブルエクスポートでは、宛先のデータベース、スキーマ、テーブルを指定する必要があります。これらの入力によって Foundry のエクスポート先が決まり、対象データベースにすでに存在する値と一致する必要があります。

宛先のデータベース、スキーマ、テーブルの値は手動で入力できます。また、探索と自動入力ボタンを使用してソースのプレビューを表示し、対象データベースに存在するテーブルを探索して、選択内容に基づいて入力欄に自動入力することもできます。

エクスポートモード

エクスポートモードを選択する必要があります。

バッチサイズ

バッチサイズとは、Foundry とソースの間でデータを転送する際に、1回のバッチで処理されるレコード数を指します。バッチサイズの調整は、データエクスポート操作のパフォーマンスと効率に影響を与え、リソース使用量の最適化と実行時間の短縮につながります。

デフォルトでは、エクスポートのバッチ処理は無効になっています。エクスポートのバッチ処理を有効にすると、2 から 5000 の間でバッチサイズを選択できます。ソースが依存するドライバーがエクスポートのバッチ処理に対応している場合、この設定は次回のエクスポート実行時に適用されます。

ヘルスチェック

ヘルスチェックはソースに対して実行され、基盤となるソースの稼働状態と可用性を確認します。デフォルトでは60分ごとに実行され、ソースに割り当てられた各 Agent から実行されます。JDBC ソースでは、ヘルスチェックは SELECT 1 クエリとして実装されています。

エクスポートタスク（レガシー）

対応データベース

エクスポートタスクは、次のデータベースに対応しています。

• PostgreSQL
• Microsoft SQL Server
• MySQL
• IBM DB2
• Oracle
• Teradata
• Snowflake
• Vertica

タスクの設定

基本的な JDBC エクスポートタスクは次のようになります。

Copied!
1
2
3
4
5
6
7
8
9
10
type: export-jdbc-task
exporterType: parquet
datasets:
  - inputAlias: myInput  # 入力が2個以上ある場合のみ必須
    writeDirectly: true
    writeMode: Overwrite
    table:
      database: mydb  # 任意
      schema: public  # 任意
      table: mytable

設定オプションの全体は次のとおりです。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
type: export-jdbc-task
exporterType: parquet

# 並列実行
parallelize: false  # データセットごとのエクスポートを並列で実行します（最大8個）

# SQL 実行フック
preSql:  # エクスポート前に実行される SQL 文
  - TRUNCATE TABLE staging_table
stagingSql:  # テーブル名の変更前に実行される SQL 文
  - SELECT 1
afterSql:  # エクスポート後に実行される SQL 文
  - INSERT INTO production_table SELECT * FROM staging_table

# トランザクション管理
manualTransactionManagement: false  # トランザクションを自分で管理するには、true に設定します
transactionIsolation: READ_COMMITTED  # java.sql.Connection.TransactionIsolationLevel

# データセット固有の設定
datasets:
  - inputAlias: myInput
    table:
      database: mydb
      schema: public
      table: mytable

    # 書き込み設定
    writeDirectly: false  # false = 最初にステージングテーブルを使用します
    copyMode: insert  # insert または directCopy（PostgreSQL のみ）
    batchSize: 1000  # INSERT 文あたりの行数
    writeMode: ErrorIfExists  # ErrorIfExists, Append, Overwrite, AppendIfPossible

    # インクリメンタル（差分）エクスポート
    incrementalType: snapshot  # snapshot または incremental

    # パフォーマンス
    exporterThreads: 1  # 並列ファイルエクスポートのスレッド数

    # 識別子の処理
    quoteIdentifiers: true  # 列名を引用符で囲みます

    # このエクスポートのトランザクション分離
    exportTransactionIsolation: READ_UNCOMMITTED

書き込みモード

• ErrorIfExists: テーブルが存在する場合は失敗します（デフォルト）
• Append: 既存のテーブルに追加します
• Overwrite: 既存のテーブルデータを置き換えます
• AppendIfPossible: 初回のエクスポートでは上書きし、それ以降は追加します。インクリメンタルエクスポートを実行する場合に適しています。

直接書き込みとステージング

writeDirectly: false（デフォルト）の場合、書き込みの流れは次のとおりです。

1. データを {tableName}_staged_ にエクスポートします
2. ステージングテーブルを最終テーブルの名前に変更します
3. 元のテーブルを削除します

writeDirectly: true の場合、データは最終テーブルに直接書き込まれます。これはインクリメンタルエクスポートに適している場合があります。

コピーモード

copyMode は、データベースへのデータのコピー方法を制御します。

• insert は、バッチサイズを設定できる、連続した INSERT 文（1行につき1つ）を使用します（デフォルト）
• directCopy: パフォーマンスを向上させるため、ファイルごとに1つの COPY 文を使用します。PostgreSQL データベースへのエクスポートでのみ利用可能です。

識別子の引用符囲み

quoteIdentifiers は、識別子を引用符で囲むかどうかを制御します。

デフォルトは true です。これは、特殊文字（1_column、my column、my-column）や予約語（level、values）などにより、有効な Foundry の列名がデータベースでは有効な列名にならない場合が多いためです。

Oracle DB または IBM DB2 にエクスポートする場合は、この設定を false にすることを検討してください。これらのデータベースでは、列を引用符で囲むと、その列を参照するたびに引用符で囲む必要があるためです。

インクリメンタル（差分）エクスポート

インクリメンタルエクスポートでは、前回のエクスポート以降に変更されたデータのみをエクスポートできます。incrementalType を incremental に設定すると、初回のエクスポートはスナップショットと同様に動作し、すべてのデータをエクスポートして、最後にエクスポートしたトランザクションを記憶します。それ以降のエクスポートでは、前回のトランザクションがデータセット内にまだ存在する場合にのみ、新しいトランザクションだけが含まれます。存在しない場合は、完全なスナップショットに戻ります。

JDBC エクスポートでは、このオプションは全体ではなくデータセットごとに指定します。

推奨される方法は、ステージングテーブルのパターンを使用することです。インクリメンタルデータを一時テーブルに書き込み、preSql 文と afterSql 文を使用して最終的な宛先にマージします。追加の一時テーブルが作成されないように writeDirectly: true を設定し、初回の実行では上書き、それ以降の実行では追加するように writeMode: AppendIfPossible を使用します。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
type: export-jdbc-task
preSql:
  - TRUNCATE TABLE mytable_incremental
datasets:
  - inputAlias: myInput
    table:
      schema: public
      table: mytable_incremental
    writeDirectly: true
    writeMode: AppendIfPossible
    incrementalType: incremental
afterSql:
  - INSERT INTO mytable SELECT * FROM mytable_incremental

データベース固有の考慮事項

Teradata

エクスポートでは、データセットの最初の列をパーティションキーとして扱う MULTISET テーブルが作成されます。行が1つのパーティション内でインデックス化されないように、最初の列に多くの値が含まれるよう列の順序を変更してください。

Teradata は、1つのトランザクション内で複数のデータ定義言語の文を実行することに対応していません。エクスポーターは、DROP、CREATE、RENAME のそれぞれの後にコミットを実行します。

デフォルトでは、Foundry の文字列列は LONG VARCHAR としてエクスポートされます。デフォルトをオーバーライドして代わりに VARCHAR(N) 型を使用するには、次のように dbOptions 設定を使用して列ごとに varchar サイズを指定できます。

Copied!
1
2
3
4
5
6
7
datasets:
  - table:
      table: mytable
    dbOptions:
      type: 'teradata'
      columnVarcharSize:
        someColumnName: 1000

Snowflake

Snowflake にエクスポートする場合は、batchSize を約50,000に増やすことを推奨します。これにより、Agent のヒープメモリーの使用量が増えることに留意してください。

Oracle

Oracle データベースにエクスポートする場合は、 columnVarcharSize を使用して列の最大サイズ（デフォルトは255バイト）を調整できます。

Copied!
1
2
3
4
5
6
7
datasets:
  - table:
      table: mytable
    dbOptions:
      type: 'oracle'
      columnVarcharSize:
        someColumnName: 1000

パフォーマンスの最適化

大量のデータをエクスポートする場合は、次のオプションの使用を検討してください。

• parallelize: true を設定すると、各入力データセットのエクスポートが並列で実行されます。スレッド数は datasets.[].exporterThreads で定義します。このデータセット内の各ファイルのエクスポートは、指定したスレッド数で並列に実行されます。最初に設定する値としては、Agent で利用可能なコア数が適しています。

• データベースとメモリーの制約に基づいて batchSize を調整します

トラブルシューティング

バッチ挿入エラー（PreparedStatement batch request の失敗など）が発生した場合は、次の解決策を検討してください。

列の順序： カーディナリティの高い列をデータセットの先頭に配置します。Teradata などのデータベースでは最初の列をパーティションキーとして使用するため、最初の列のカーディナリティが低いと、すべての行が単一のパーティション内でインデックス化され、失敗につながることがあります。

文字列のエンコーディング： 文字列データが、対象データベースに対応する文字エンコーディングを使用していることを確認してください。Teradata では UTF-8 ではなく、ASCII または Latin-1 エンコーディングが必要です。エクスポート前にデータをクレンジングまたは変換する必要がある場合があります。

バッチサイズによるデバッグ： batchSize を1に減らして、問題のある行を特定します。これによりエクスポートは遅くなりますが、データ品質の問題や制約違反によって失敗を引き起こしている行を正確に特定しやすくなります。