Redshift バッチソース

このページでは、Cloud Data Fusion で Amazon Redshift インスタンスから Google Cloud にデータを読み込む方法について説明します。Redshift ソースコネクタによって、Redshift データセットから BigQuery などの宛先にテーブルを同期できます。このコネクタによって、構成可能な SQL クエリを作成することもできます。

始める前に

Cloud Data Fusion バージョン 6.9.0 以降は、Redshift ソースをサポートしています。
Redshift ソースコネクタを構成する場合は、再利用可能な既存の接続を選択するか、新しい 1 回限りの接続を作成できます。詳細については、接続を管理するをご覧ください。接続を再利用する場合は、次の点に注意してください。
- 認証情報を指定する必要はありません。
- 既存の接続には、インポートクエリの生成に使用されるスキーマとテーブル名の情報が用意されています。

プラグインを構成する

Cloud Data Fusion ウェブインターフェースに移動し、[Studio] をクリックします。
[Realtime] ではなく、[Data Pipeline - Batch] が選択されていることを確認します。
[ソース] メニューで [Redshift] をクリックします。Redshift ノードがパイプラインに表示されます。[Studio] ページに Redshift ソースが表示されない場合は、Cloud Data Fusion ハブから Redshift ソースコネクタをデプロイします。
ソースを構成するには、Redshift ノードに移動して [プロパティ] をクリックします。
次のプロパティを入力します。完全版リストについては、プロパティをご覧ください。
1. BigQuery ノードのラベルを入力します（例: Redshift tables）。
2. 接続の詳細を入力します。新しい 1 回限りの接続または再利用可能な既存接続を設定できます。
  新しい接続
  Redshift に 1 回限りの接続を追加するには、次の手順を行います。
  1. [接続を使用] はオフのままにします。
  2. [JDBC ドライバ名] フィールドに、ドライバの名前を入力します。Redshift は、CData と Amazon の 2 種類の JDBC ドライバをサポートしています。詳細については、JDBC ドライバをアップロードするをご覧ください。
  3. [ホスト] フィールドに、Redshift クラスタのエンドポイントを入力します（例: cdf-redshift-new.example-endpoint.eu-west-1.redshift.amazonaws.com）。
  4. 省略可: [ポート] フィールドに、データベースのポート番号（例: 5439）を入力します。
  5. Redshift データベースで認証が必要な場合は、次の操作を行います。
    
    [ユーザー名] フィールドに、データベースの名前を入力します。
    
    [パスワード] フィールドに、データベースのパスワードを入力します。
    
    省略可: [引数] フィールドに、キー値引数を入力します。CData ドライバを使用するには、接続引数（RTK や OEMKey など）を指定します（該当する場合）。
    
    [名前] フィールドに名前を入力します（例: SN-PC-Source-01-01-2024）。
    
    [データベース] フィールドにターゲットデータベース名を入力します（例: datafusiondb）。
  再利用可能な接続
  既存の接続を再利用する手順は次のとおりです。
  1. [接続を使用] をオンにします。
  2. [接続を参照] をクリックします。
  3. 接続名をクリックします。
    
    注: 接続を参照するときに表示される接続の追加、インポート、編集の詳細については、接続を管理するをご覧ください。
  4. 省略可: 接続が存在せず、再利用可能な新しい接続を作成する場合は [接続を追加] をクリックし、本ページにある [新しい接続] タブの手順をご参照ください。
3. [インポートクエリ] フィールドに、Redshift ソースのスキーマ名とテーブル名を使用してクエリを入力します（例: Select * from "public"."users"）。
4. 省略可: 境界クエリや分割数などの [詳細] プロパティを入力します。すべてのプロパティの説明については、プロパティをご覧ください。
省略可: [検証] をクリックして、見つかったエラーに対処します。
[閉じる] をクリックします。プロパティが保存され、Cloud Data Fusion ウェブインターフェースでデータパイプラインの構築を続行できます。

プロパティ

プロパティ	自動化用のマクロに対応する	必要なプロパティ	説明
ラベル	×	○	データパイプラインのノードの名前。
接続を使用	×	×	ソースへの接続を参照します。[接続を使用] がオンになっている場合は、認証情報を指定する必要はありません。
接続	○	○	使用する接続の名前。 [接続を使用] が選択されている場合、このフィールドが表示されます。データベースとテーブルの情報は接続によって指定されます。
JDBC ドライバ名	○	○	使用する JDBC ドライバの名前。[接続を使用] が選択されていない場合は、このフィールドが表示されます。
ホスト	○	○	Amazon Redshift クラスタのエンドポイント。[接続を使用] が選択されていない場合は、このフィールドが表示されます。
ポート	○	×	Redshift が実行されているポート。[接続を使用] が選択されていない場合は、このフィールドが表示されます。
ユーザー名	○	×	指定したデータベースに接続するためのユーザー ID。[接続を使用] が選択されていない場合は、このフィールドが表示されます。
パスワード	○	×	指定したデータベースへの接続に使用するパスワード。[接続を使用] が選択されていない場合は、このフィールドが表示されます。
接続引数	○	×	接続引数としての任意の文字列の Key-Value ペアのリスト。これらの引数は、追加の構成が必要な可能性のある JDBC ドライバの接続引数として JDBC ドライバに渡されます。[接続を使用] が選択されていない場合は、このフィールドが表示されます。
参照名	×	○	このソースを、リネージ、メタデータのアノテーション、その他のサービスのために一意に識別します。
データベース	○	○	Redshift データベース名。データを選択するには、[データベースを参照] をクリックします。
インポートクエリ	○	○	指定したテーブルからデータをインポートするために使用する `SELECT` クエリ。
境界クエリ	○	×	`splitBy` フィールドの最小値と最大値を返す SQL クエリ。例: `SELECT MIN(id),MAX(id) FROM table`。`numSplits` が 1 に設定されている場合は必須ではありません。
Split column	○	×	スプリットの生成に使用されるフィールド名。`numSplits` が 1 に設定されている場合は必須ではありません。
分割数	○	×	生成するスプリットの数。
サイズ	○	×	分割ごとに一度にフェッチする行数。フェッチサイズを大きくすると、インポートが速くなりますが、その一方でメモリ使用量は高くなります。指定しない場合は、デフォルトの `1000` が使用されます。

データ型マッピング

次の表は、Redshift のデータ型と対応する CDAP の型を示しています。

Redshift のデータ型	CDAP スキーマのデータ型
`bigint`	`long`
`boolean`	`boolean`
`character`	`string`
`character varying`	`string`
`date`	`date`
`double precision`	`double`
`geometry`	`bytes`
`hllsketch`	`string`
`integer`	`int`
`json`	`string`
`numeric`（精度、スケール）/ `decimal`（精度、スケール）	`decimal`
精度 0 の `numeric`	`string`
`real`	`float`
`smallint`	`int`
`super`	`string`
`text`	`string`
タイムゾーンが設定されていない `time` [ (p) ]	`time`
タイムゾーンが設定されている `time` [ (p) ]	`string`
タイムゾーンが設定されていない `timestamp` [ (p) ]	`timestamp`
タイムゾーンが設定されている `timestamp` [ (p) ]	`timestamp`
`varbyte`	`byte`
`xml`	`string`

ベストプラクティス

Google Cloud から Redshift クラスタに接続する場合は、次のベストプラクティスを適用します。

IP アドレス許可リストを使用する

未承認のソースからのアクセスを防止し、特定の IP アドレスへのアクセスを制限するには、Redshift クラスタでアクセス制御を有効にします。

Redshift アクセス制御を使用している場合、Cloud Data Fusion でクラスタにアクセスする手順は次のとおりです。

Redshift クラスタに接続する必要がある Google Cloud のサービスまたはマシンの外部 IP アドレス（プロキシサーバー IP など）を取得します（IP アドレスの表示をご覧ください）。Dataproc クラスタの場合は、すべてのマスターノードと子ノードの IP アドレスを取得します。
Google Cloud マシンの IP アドレスの受信ルールを作成して、セキュリティグループの許可リストに IP アドレスを追加します。
Wrangler で接続プロパティを追加してテストします。
1. ウェブインターフェースで Cloud Data Fusion インスタンスを開きます。
2. [Wrangler] > [接続を追加] をクリックして、Redshift の新しい接続を作成します。
3. すべての接続プロパティを入力します。
4. [テスト接続] をクリックして、問題を解決します。

複数の分割を作成するには、境界クエリを使用します。

複数の分割の場合は、境界クエリを使用してマルチノードクラスタを管理します。Redshift からデータを抽出し、負荷を各ノードに均一に負荷を分散するシナリオでは、Redshift ソースコネクタのプロパティに境界クエリを構成します。

Cloud Data Fusion パイプラインの [Studio] ページで、Redshift ノードに移動し、[プロパティ] をクリックします。
[詳細] プロパティで、次のことを指定します。
1. 作成するスプリットの数を入力します。
2. 各スプリットのフェッチサイズを入力します。
3. マルチノード Redshift クラスタに適用する境界クエリを入力します。
4. [分割列] フィールド名に入力します。

たとえば、次のようなユースケースがあるとします。

1,000 万のレコードを含むテーブルがあります。
id という一意の ID 列があります。
Redshift クラスタには 4 つのノードがあります。
目的: クラスタの可能性を最大限に活用するために、複数の分割を生成します。これを実施するには、次のプロパティ構成を使用します。
- [境界クエリ] フィールドに、次のクエリを入力します。
```
SELECT MIN(id), MAX(id) FROM tableName
```
  このクエリで、id はスプリットが適用される列の名前です。
- [分割列] フィールドに列名の id を入力します。
- スプリット数とフェッチサイズを入力します。これらのプロパティは相互に関連しているため、フェッチサイズに基づいてスプリットを計算することも、その逆も可能です。この例では、次のものを入力します。
  
  重要: フェッチサイズが大きいと、インポートを速くできます。トレードオフとして、メモリ使用量が増加します。
- [スプリット数] フィールドに 40 と入力します。この例では、テーブルに 1,000 万のレコードがあり、40 のスプリットを作成すると、各スプリットには 25 万のレコードが含まれます。
- [フェッチサイズ] フィールドに 250,000 と入力します。

次のステップ

Cloud Data Fusion のプラグインを確認する。