DataHub データを Hologres に同期する - Hologres - Alibaba Cloud ドキュメントセンター

DataHub から Hologres へのデータ同期は、リアルタイムデータ分析、リアルタイムモニタリング、リアルタイムレポート、および複雑なデータ構造処理を必要とするシナリオに最適です。DataHub と Hologres の機能を組み合わせることで、データ処理と分析の効率と精度が向上します。このトピックでは、DataHub データ同期タスクを作成する方法と、よくある質問 (FAQ) への回答について説明します。

背景情報

DataHub には、データシンクおよびソース機能を備えたデータ同期機能があります。この機能を使用して、Topic のデータを他の Alibaba Cloud プロダクトにreal-time or Near Real-Timeで同期でき、プロダクト間のデータフローを実現します。DataHub のデータ同期機能の詳細については、「概要」をご参照ください。

次の表は、DataHub と Hologres 間のデータの型マッピングについて説明します。

DataHub	Hologres
プロジェクト	データベース
Topic	テーブル

同期シナリオと戦略

同期シナリオ

同期シナリオ	説明
行単位の挿入	ソースデータベースからターゲットデータベースに行単位でデータを挿入します。これは、すべての DataHub データを Hologres に同期するシナリオに適用されます。
再生	バイナリログ (Binlog) データベース操作ログを分析して再実行し、ソースデータベースの変更をターゲットデータベースに適用して、データ整合性を確保します。これは、DTS が DataHub にデータを同期し、その後 DataHub データが Hologres に同期され、DataHub が Binlog として機能するシナリオに適用されます。説明 DTS が DataHub にデータを同期する場合、操作情報を記録するためにデータ列に補助列を追加します。DTS 同期から DataHub へのこれらの補助列の命名は、古いルールと新しいルールで異なります。詳細については、「データ同期の補助列ルールを変更する」をご参照ください。

同期戦略

同期戦略	説明
上書き (置き換え)	データ書き込み中にプライマリキーの競合が発生した場合、新しいデータが古いデータに上書きされ、書き込まれることで、ターゲットデータベースとソースデータベース間のデータ整合性が確保されます。
無視 (ignore)	データ書き込み中にプライマリキーの競合が発生した場合、新しいデータは無視されます。これは、データが更新されず、重複データのインポートと上書きを防ぎ、ターゲットデータベースのデータ整合性を維持することを意味します。

注意事項

DataHub から Hologres には TUPLE 型のデータのみを同期できます。
パーティションテーブルにデータを書き込む前に、Hologres で子テーブルを作成する必要があります。詳細については、「CREATE PARTITION TABLE」をご参照ください。
各同期タスクは特定の数の接続を消費します。タスクが消費する接続数は、DataHub Topic のシャード数に等しくなります。

事前準備

DataHub サービスを有効にし、DataHub データを準備します。詳細については、「クイックスタート (同期例)」をご参照ください。
Hologres インスタンスを購入し、テーブルを作成します。このトピックでは、lineitem テーブルを例として使用します。詳細については、「Hologres インスタンスの購入」および「HoloWeb を使用した Hologres でのテーブル作成」をご参照ください。
次の表は、DataHub と Hologres 間のデータの型マッピングについて説明します。
DataHub
Hologres
TINYINT
SMALLINT
SMALLINT
SMALLINT
INTEGER
INTEGER
BIGINT
BIGINT
FLOAT
REAL
DOUBLE
DOUBLE PRECISION
DECIMAL
DECIMAL
STRING
TEXT
BOOLEAN
BOOLEAN
TIMESTAMP
TIMESTAMPTZ
以下は、テーブル作成ステートメントの例です。
```
BEGIN;
CREATE TABLE lineitem ( 
L_ORDERKEY BIGINT NOT NULL,
L_PARTKEY BIGINT NOT NULL,
L_SUPPKEY BIGINT NOT NULL,
L_LINENUMBER BIGINT NOT NULL,
L_QUANTITY DECIMAL(20,10),
L_EXTENDEDPRICE DECIMAL(20,10),
L_DISCOUNT DECIMAL(20,10),
L_TAX DECIMAL(20,10),
L_RETURNFLAG TEXT,
L_LINESTATUS TEXT,
L_SHIPDATE TIMESTAMPTZ,
L_COMMITDATE TIMESTAMPTZ,
L_RECEIPTDATE TIMESTAMPTZ,
L_SHIPINSTRUCT TEXT,
L_SHIPMODE TEXT,
L_COMMENT TEXT
);
CALL set_table_property('lineitem', 'orientation', 'column');
COMMIT;
```

同期タスクの作成

DataHub サービスコンソールにログインします。作成した Topic をクリックして、Topic 詳細ページに移動します。
Topic 詳細ページの右上隅で、[+Sync] をクリックします。

Hologres をクリックし、[コネクタの作成] ページでパラメーターを設定します。

パラメーター	説明
インスタンス	Hologres のインスタンス ID です。インスタンス ID を取得するには、Hologres 管理コンソールに移動します。
Database	DataHub データを受信する Hologres データベースの名前。
テーブル	DataHub データを受信する Hologres テーブル `lineitem` の名前。
プライマリキー競合ポリシー	プライマリキーの競合が発生した場合のデータ更新ポリシー。値は次のとおりです。 replace (デフォルト): データ書き込み中にプライマリキーの競合が発生した場合、新しいデータが古いデータに上書きされます。 ignore: データ書き込み中にプライマリキーの競合が発生した場合、新しいデータは無視されます。これは、データが更新されず、古いデータが保持されることを意味します。データ同期戦略の詳細については、「同期戦略」をご参照ください。
[同期シナリオ]	データ同期のユースケース。値は次のとおりです。 default (デフォルト): 行単位の挿入。 dts: DTS が DataHub にデータを同期するときに、新しい補助列ルールを有効にする場合は、このパラメーターを選択します。 dts_old: DTS が DataHub にデータを同期するときに、新しい補助列ルールを有効にしない場合は、このパラメーターを選択します。データ同期シナリオの詳細については、「同期シナリオ」をご参照ください。
[インポートフィールド]	Hologres にインポートするフィールド。必要に応じて、一部またはすべてのフィールドをインポートします。
[認証モード]	デフォルトは AccessKey です。
[AccessKey ID]	Hologres インスタンスにアクセスするための AccessKey ID。ユーザーの AccessKey ID を取得するには、「AccessKey 管理」をクリックします。
[AccessKey Secret]	Hologres インスタンスにアクセスするための AccessKey Secret。AccessKey Secret を取得するには、「AccessKey 管理」をクリックします。
タイムスタンプ単位	同期時間単位。値は次のとおりです。 MICROSECOND: マイクロ秒 (デフォルト値)。 MILLISECOND: ミリ秒。第二: 第二。

Create をクリックして、DataHub から Hologres にデータを同期します。
コネクタを作成すると、Topic 詳細ページの [Sync Tasks] タブで、リアルタイムデータ同期ステータスを表示できます。
Hologres でデータをクエリします。
Hologres インスタンスの開発ツールに接続して、Hologres にリアルタイムで同期されたデータをクエリします。Hologres への接続方法の詳細については、「Hologres に接続する」をご参照ください。次のサンプルクエリステートメントを実行できます。
```
SELECT COUNT(*) FROM lineitem;
```

よくある質問

このセクションでは、Hologres の使用時に発生する可能性のある一般的なエラーと、問題のトラブルシューティングと解決に役立つソリューションについて説明します。

質問 1
- エラーメッセージ
```
ErrorMessage: Import field not found in dest schema.
```
- 原因
  - 同期タスクで指定されたインポートフィールドが Hologres テーブルに存在しません。
  - タスクの同期シナリオがデフォルトに設定されていますが、インポートフィールドには、DTS が DataHub にデータを同期するときに生成される補助列が含まれています。
- ソリューション
  - Hologres テーブルを再作成し、不足しているインポートフィールドを追加します。または、同期タスクを変更して、Hologres テーブルに存在しないインポートフィールドを削除することもできます。
  - 同期タスクを再作成し、同期シナリオを dts または dts_old に設定します。

質問 2
- エラーメッセージ
```
ErrorMessage: Column type not match with Holo column.
```
- 原因
  DataHub Topic のフィールドのデータの型が、Hologres テーブルの対応するフィールドのデータの型と一致しません。
- ソリューション
  データの型マッピングに基づいて、正しいフィールドの型で Hologres テーブルを再作成します。
質問 3
- エラーメッセージ
```
ErrorMessage: Not import column xxx not allow null and no default value.
```
- 原因
  Hologres テーブルの一部のフィールドは同期タスクのインポートフィールドに含まれていません。ただし、これらのフィールドは NOT NULL プロパティで構成されており、デフォルト値がありません。
- ソリューション
  Hologres テーブルを再作成します。同期タスクのインポートフィールドに含まれていないフィールドについては、`NOT NULL` プロパティを設定しないか、デフォルト値を設定しないでください。

DataHub	Hologres
TINYINT	SMALLINT
SMALLINT	SMALLINT
INTEGER	INTEGER
BIGINT	BIGINT
FLOAT	REAL
DOUBLE	DOUBLE PRECISION
DECIMAL	DECIMAL
STRING	TEXT
BOOLEAN	BOOLEAN
TIMESTAMP	TIMESTAMPTZ