MaxCompute へのデータ同期

準備

1. MaxCompute テーブルを作成する

DataHub では、MaxCompute テーブルにデータを同期できます。パーティションテーブルと非パーティションテーブルの両方がサポートされています。MaxCompute でのデータ処理を容易にするために、パーティションテーブルにデータを同期することをお勧めします。

DataHub では、TUPLE 型または BLOB 型のトピックから MaxCompute テーブルにデータを同期できます。

(1) TUPLE 型のトピックからデータを同期する場合、MaxCompute テーブルのデータ型は DataHub トピックのデータ型と一致する必要があります。次の表に、データ型のマッピングを示します。

MaxCompute	DataHub
BIGINT	BIGINT
STRING	STRING
BOOLEAN	BOOLEAN
DOUBLE	DOUBLE
DATETIME	TIMESTAMP
DECIMAL	DECIMAL
TINYINT	TINIINT
SMALLINT	SMALLINT
INT	INTEGER
FLOAT	FLOAT
MAP	サポートされていません
ARRAY	サポートされていません

DataHub は MaxCompute のすべてのデータ型をサポートしているわけではありません。DataHub でサポートされているデータ型に基づいて MaxCompute スキーマを作成することをお勧めします。

(2) BLOB 型のトピックからデータを同期する場合、MaxCompute スキーマには STRING 型のフィールドが 1 つだけ含まれている必要があります。デフォルトでは、DataHub はこのフィールドにデータを同期します。

DataHub	MaxCompute
BLOB	STRING

(3) データの追跡とトラブルシューティングを容易にするために、MaxCompute スキーマを作成するときに、SRTING 型の __rowkey__ フィールドを追加することをお勧めします。DataHub は、DataHub 内の同期済みデータのトレース情報をこのフィールドに自動的に同期し、後続のデータのトラブルシューティングを容易にします。

2. 必要な権限が付与されたデータ同期用のアカウントを準備する

(1) MaxCompute にデータを同期するための DataConnector を作成する場合、MaxCompute テーブルにアクセスするために使用されるアカウント情報を手動で入力する必要があります。入力したアカウント情報が有効であることを確認してください。ほとんどの場合、MaxCompute に RAM ユーザーを使用できます。

(2) アカウントに MaxCompute テーブルに対する必要な権限を付与する必要があります。権限には、CreateInstance、Describe、Alter、および Update 権限が含まれます。

DataWorks コンソールで MaxCompute テーブルの権限を管理できます。詳細については、「MaxCompute エンジン権限を構成する」をご参照ください。MaxCompute のコマンドラインツールを権限付与に使用することもできます。

3. TimestampUnit パラメーターの値を決定する。

(1) TimestampUnit パラメーターは、TIMESTAMP 型のデータが変換され、MaxCompute の DATETIME 型のフィールドに書き込まれる単位を指定します。

(2) TIMESTAMP 型のフィールドを秒単位で MaxCompute に書き込む場合は、DataConnector を作成するときに TimestampUnit ドロップダウンリストから SECOND を選択します。TIMESTAMP 型のフィールドをミリ秒単位で MaxCompute に書き込む場合は、MILLISECOND を選択します。TIMESTAMP 型のフィールドをマイクロ秒単位で MaxCompute に書き込む場合は、MICROSECOND を選択します。

使用上の注意

MaxCompute の現在の書き込み標準のため、パーティションが多いほど DataHub からのデータ同期が遅くなります。MaxCompute にデータを同期するための DataConnector を作成する場合は、特に USER_DEFINE モードでは、パーティションの数を制限することをお勧めします。
- シャード内のレコードのタイムスタンプとシャードができるだけ連続していることを確認してください。
- すべてのパーティションモードでパーティションを作成するときに、過剰なパーティションを作成しないでください。

DataConnector を作成する

DataHub コンソールの左側のナビゲーションウィンドウで、[プロジェクトマネージャー] をクリックします。[プロジェクトリスト] ページで、プロジェクトを見つけて、[アクション] 列の [表示] をクリックします。プロジェクトの詳細ページで、トピックを見つけて、[アクション] 列の [表示] をクリックします。
右上隅にある同期 +Connector ボタンをクリックして、同期タスクを作成します
[コネクタの作成] パネルで、次の図に示すように、[MaxCompute] をクリックします。
(2) BLOB 型のトピックからデータを同期する

odps_2

いくつかのパラメーターの説明:

次の部分では、DataHub コンソールで DataConnector を作成するために使用されるいくつかのパラメーターについて説明します。同期構成の詳細については、DataHub SDK の説明をご参照ください。

インポートフィールド
DataHub は、設定に基づいて特定のフィールドの値を MaxCompute テーブルに同期できます。
パーティションモード
パーティションモードは、MaxCompute のどのパーティションにデータが書き込まれるかを決定します。次の表に、DataHub でサポートされているパーティションモードを示します。

パーティションモード	パーティション基準	サポートされているトピックのデータ型	説明
USER_DEFINE	レコード内の MaxCompute のパーティションフィールドと同じ名前を持つパーティションフィールドの値	TUPLE	(1). DataHub スキーマには、MaxCompute のパーティションフィールドが含まれている必要があります。(2) このフィールドの値は `UTF-8 文字列`である必要があります。このフィールドの値は空にすることができ、これはデータがパーティション化されていないことを示します。
SYSTEM_TIME	レコードが DataHub に書き込まれた時刻	TUPLE または BLOB	(1). パーティション構成で MaxCompute パーティションの時刻形式を設定します。(2) タイムゾーンを指定します。
EVENT_TIME	レコード内の `event_time(TIMESTAMP)` フィールドの値	TUPLE	(1). パーティション構成で MaxCompute パーティションの時刻形式を設定します。(2) タイムゾーンを指定します。
META_TIME	レコード内の `__dh_meta_time__` フィールドの値	TUPLE / BLOB	(1). パーティション構成で MaxCompute パーティションの時刻形式を設定します。(2) タイムゾーンを指定します。

SYSTEM_TIME、EVENT_TIME、および META_TIME は、同期されるデータがタイムスタンプとタイムゾーンの構成に基づいてパーティション化されるパーティションモードです。デフォルトの単位はマイクロ秒です。

パーティション構成: 同期されるデータがパーティション化されるタイムスタンプ構成。DataHub は MaxCompute の固定時刻形式をサポートしています。次の表に、固定時刻形式と対応するパーティションを示します。

パーティション	時刻形式	説明
ds	%Y%m%d	日
hh	%H	時
mm	%M	分

時間範囲: データがタイムスタンプに基づいてパーティション化されるときのパーティションの間隔。有効な値: 15 ～ 1440。単位: 分。値 1440 は 1 日を示します。パラメーターのステップサイズは 15 分です。
タイムゾーン: データがパーティション化され、MaxCompute に書き込まれるタイムスタンプのタイムゾーン。
分割キー: BLOB 型のデータを同期するための DataConnector を作成する場合、16 進数のデリミタを指定して BLOB 型のデータを分割し、MaxCompute に同期できます。たとえば、0A は 改行 (\n) を示します
Base64 エンコード: デフォルトでは、DataHub の BLOB 型のデータはバイナリデータであり、MaxCompute のマップされたフィールドは STRING 型です。したがって、DataHub コンソールで DataConnector を作成する場合、デフォルトでは同期前に Base64 エンコーディングが実行されます。DataHub SDK を使用して、より多くのカスタマイズ要件を満たすことができます。

DataConnector を表示する

odps_3

例

1. USER_DEFINE モード

DataHub トピックを作成します。
注: トピックのスキーマには、MaxCompute のパーティションフィールドが含まれている必要があり、フィールドは STRING 型です (次の図を参照)。
DataHub トピックにデータを書き込みます。DataHub SDK を使用してデータを書き込むことができます。
テスト中は、DataHub SDK を使用して複数のレコードを書き込みます。 [ds,hh,mm] の値は [20210304,01,15] と [20210304,02,15] です。次の図はレコードを示しています。

mc_02

3. DataConnector を作成します。

USER_DEFINE モードでは、DataConnector を作成するときにパーティションフィールドを構成できます。使用可能な MaxCompute テーブルが存在しない場合は、テーブルが自動的に作成されます。

mc_03

この例では、インポートされるフィールドは f1 フィールドと f2 フィールドです。 f3 フィールドは同期されません。

4. 同期されたデータを確認します。

odps_4

MaxCompute テーブルのデータをクエリします。次の図はクエリ結果を示しています。 odps+5

USER_DEFINE モードでは、DataHub は DataHub から MaxCompute のパーティションフィールドの値に基づいて対応するパーティションにデータを同期します。

2. SYSTEM_TIME モード

DataHub トピックを作成します。
注: データが書き込まれるパーティションは、データが DataHub に書き込まれた時刻に基づいて決定されます。したがって、トピックスキーマにはデータフィールドのみが含まれている必要があり、パーティションフィールドを含める必要はありません (次の図を参照)。
DataHub トピックにデータを書き込みます。DataHub SDK を使用してデータを書き込むことができます。
テスト中は、DataHub SDK を使用して複数のレコードを書き込みます。データが DataHub に書き込まれた時刻は 2021-03-04 14:02:45 です。次の図はレコードを示しています。
DataConnector を作成します。

4. 同期されたデータを確認します。

DataHub コンソールで DataConnector の同期情報 (DoneTime など) を表示できます。

mc_08

MaxCompute テーブルのデータをクエリします。次の図はクエリ結果を示しています。

mc_09

SYSTEM_TIME モードでは、DataHub は DataHub から データが DataHub に書き込まれた時刻に基づいて対応するパーティションにデータを同期します。

FAQ

問題: MaxCompute へのデータ同期のタイムスタンプが 1970-01-19 に変更されます。
原因: DataHub から MaxCompute へのデータ同期のタイムスタンプのデフォルトの単位はマイクロ秒ですが、データが DataHub に書き込まれたときのタイムスタンプの単位はミリ秒です。解決策: データを DataHub に書き込むときに、タイムスタンプの単位にマイクロ秒を使用します。