DataWorks:PolarDB データソース

バッチデータの読み取りと書き込み

ビューのデータを読み取ることができます。

リアルタイムデータの読み取り

同期タスクのソースが PolarDB for MySQL クラスタの場合、クラスタのバイナリロギング機能を有効にする必要があります。 PolarDB for MySQL は MySQL と完全に互換性があり、高レベルの物理ログを使用してバイナリログを置き換えます。 PolarDB と MySQL エコシステム間の統合を容易にするために、PolarDB クラスタのバイナリロギング機能を有効にすることができます。

IP アドレスホワイトリストを設定する

Data Integration 専用リソースグループが存在する仮想プライベートクラウド (VPC) の CIDR ブロックを、PolarDB for MySQL クラスタの IP アドレスホワイトリストに追加する必要があります。

バイナリロギング機能を有効にする

詳細については、「バイナリロギングを有効にする」をご参照ください。

データソースを追加する

DataWorks で同期タスクを開発する前に、「データソース管理」の手順に従って、必要なデータソースを DataWorks に追加する必要があります。 データソースを追加する際に、DataWorks コンソールのパラメータのヒントを参照して、パラメータの意味を理解することができます。

単一テーブルのデータを同期するためのバッチ同期タスクを設定する

• 設定手順の詳細については、「コーディングレス UI を使用してバッチ同期タスクを設定する」および「コードエディタを使用してバッチ同期タスクを設定する」をご参照ください。

• コードエディタを使用してバッチ同期タスクを設定する際に設定されるすべてのパラメータと実行されるコードについては、「付録：コードとパラメータ」をご参照ください。

単一テーブルのデータまたはデータベースのすべてのデータを同期するためのリアルタイム同期タスクを設定する

設定手順の詳細については、「DataStudio でリアルタイム同期タスクを設定する」をご参照ください。

データベース内のすべてのデータのバッチ同期と、単一テーブルのデータまたはデータベース内の完全データまたは増分データのリアルタイム同期を実装するための同期設定を行う

設定手順の詳細については、「Data Integration で同期タスクを設定する」をご参照ください。

コードエディタを使用してバッチ同期タスクを設定する

コードエディタを使用してバッチ同期タスクを設定する場合は、統一スクリプト形式の要件に基づいて、スクリプトに関連パラメータを設定する必要があります。 詳細については、「コードエディタを使用してバッチ同期タスクを設定する」をご参照ください。 次の情報では、コードエディタを使用してバッチ同期タスクを設定する際にデータソースに設定する必要があるパラメータについて説明します。

PolarDB Reader のコードのパラメータ

PolarDB Writer のコードのパラメータ

• PolarDB Writer のコードのパラメータ

  パラメータ	説明	必須	デフォルト値
  datasource	データソースの名前。 追加されたデータソースの名前と同じである必要があります。 コードエディタを使用してデータソースを追加できます。	はい	デフォルト値なし
  table	データを読み取るテーブルの名前。	はい	デフォルト値なし
  writeMode	書き込みモード。 有効値： insert: このモードは、コーディングレス UI の INSERT INTO と同等です。 update (コーディングレス UI の ON DUPLICATE KEY UPDATE と同等) replace: このモードは、コーディングレス UI の REPLACE INTO と同等です。 書き込みモードとデータ例の詳細については、writeMode パラメータの説明 を参照してください。 説明 デスティネーションが PolarDB for PostgreSQL の場合、insert モードのみがサポートされます。 データを更新し、プライマリキーの競合を回避するには、バッチ同期を実行する前に、デスティネーションテーブル内の重複レコードを削除することをお勧めします。 次のアプローチのいずれかを使用できます。 アプローチ 1：preSql パラメータ (コーディングレス UI の デスティネーションにデータを書き込む前に実行される文 と同等) に TRUNCATE 文を設定して、デスティネーションテーブルをクリアします。 アプローチ 2：バッチ同期ノードのアップストリームノードでデスティネーションテーブルを処理して、データ同期中にプライマリキーの競合が発生しないようにします。	いいえ	insert
  column	データを書き込む列の名前。 名前をコンマ (,) で区切ります。 例："column":["id","name","age"]。 デスティネーションテーブルのすべての列にデータを書き込む場合は、このパラメータをアスタリスク (*) に設定します (例："column":["*"])。	はい	デフォルト値なし
  preSql	バッチ同期タスクの実行前に実行する SQL 文。 たとえば、このパラメータを古いデータを削除するために使用される SQL 文に設定できます。 コーディングレス UI では 1 つの SQL 文のみを実行でき、コードエディタでは複数の SQL 文を実行できます。	いいえ	デフォルト値なし
  postSql	バッチ同期タスクの実行後に実行する SQL 文。 たとえば、このパラメータをタイムスタンプを追加するために使用される SQL 文に設定できます。 コーディングレス UI では 1 つの SQL 文のみを実行でき、コードエディタでは複数の SQL 文を実行できます。	いいえ	デフォルト値なし
  batchSize	一度に書き込むデータレコードの数。 ビジネス要件に基づいて、このパラメータを適切な値に設定します。 これにより、Data Integration と PolarDB 間の相互作用が大幅に削減され、スループットが向上します。 このパラメータを過度に大きい値に設定すると、データ同期中にメモリ不足 (OOM) エラーが発生する可能性があります。	いいえ	1,024
  updateColumn	プライマリキーの競合または一意なインデックスの競合が発生した場合に更新される列の名前。 このパラメータは、writeMode パラメータが update に設定されている場合にのみ有効になります。 "updateColumn": ["name", "age"] など、名前をコンマ (,) で区切ります。 説明 PolarDB for MySQL データソースのみがこのパラメータをサポートしています。	いいえ	デフォルト値なし

• writeMode パラメータの説明

  項目	insert (コーディングレス UI の INSERT INTO と同等)	update (コーディングレス UI の ON DUPLICATE KEY UPDATE と同等)	replace (コーディングレス UI の REPLACE INTO と同等)
  処理ルール	プライマリキーの競合または一意なインデックスの競合が発生した場合、データは競合する行に書き込まれず、これらの行に書き込まれないデータはダーティデータと見なされます。	プライマリキーの競合または一意なインデックスの競合が発生しない場合、データは、このパラメータを insert に設定した場合と同じ方法で処理されます。 競合が発生した場合、デスティネーションテーブル内の競合する行のデータは新しいデータに置き換えられます。	プライマリキーの競合または一意なインデックスの競合が発生しない場合、データは、このパラメータを insert に設定した場合と同じ方法で処理されます。 競合が発生した場合、元の行は削除され、新しい行が挿入されます。 これは、元の行のすべてのフィールドが置き換えられることを示します。
  データ例	ソーステーブルのデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ デスティネーションテーブルの元のデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | | +----+---------+-----+ バッチ同期タスクの実行後、1 つのデータレコードがデスティネーションテーブルに書き込まれ、1 つのデータレコードが生成されます。 +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | wangwu | | +----+---------+-----+	シナリオ 1：一部の列のみが指定されている："column": ["id","name"]。 ソーステーブルのデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ デスティネーションテーブルの元のデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ バッチ同期タスクの実行後、2 つのデータレコードがデスティネーションテーブルに書き込まれ、ダーティデータレコードは生成されません。 +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | 3 | +----+---------+-----+ シナリオ 2：すべての列が指定されている："column": ["id","name","age"]。 ソーステーブルのデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ デスティネーションテーブルの元のデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ バッチ同期タスクの実行後、2 つのデータレコードがデスティネーションテーブルに書き込まれ、ダーティデータレコードは生成されません。 +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+	ソーステーブルのデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+ デスティネーションテーブルの元のデータ +----+---------+-----+ | id | name | age | +----+---------+-----+ | 2 | wangwu | 3 | +----+---------+-----+ バッチ同期タスクの実行後、2 つのデータレコードがデスティネーションテーブルに書き込まれ、ダーティデータレコードは生成されません。 +----+---------+-----+ | id | name | age | +----+---------+-----+ | 1 | zhangsan| 1 | | 2 | lisi | | +----+---------+-----+