すべてのプロダクト
Search

このプロダクト

    DataHub:MaxCompute 同期コネクタの作成

    ドキュメントセンター

    DataHub:MaxCompute 同期コネクタの作成

    最終更新日:Aug 26, 2026

    前提条件

    MaxCompute テーブルの作成

    パーティションテーブルを使用して、MaxCompute でのデータ処理を簡素化します。

    DataHub は、TUPLE トピックおよび BLOB トピックから MaxCompute テーブルへのデータ同期をサポートしています。

    • TUPLE トピックの場合、ターゲットの MaxCompute テーブルのデータ型は、DataHub トピックのスキーマと互換性がある必要があります。次の表に、データ型のマッピングを示します。

      MaxCompute

      DataHub

      BIGINT

      BIGINT

      STRING

      STRING

      BOOLEAN

      BOOLEAN

      DOUBLE

      DOUBLE

      DATETIME

      TIMESTAMP

      DECIMAL

      DECIMAL

      TINYINT

      TINYINT

      SMALLINT

      SMALLINT

      INT

      INTEGER

      FLOAT

      FLOAT

      MAP

      サポートされていません

      ARRAY

      サポートされていません

      MaxCompute テーブルスキーマを作成するときは、DataHub がサポートするデータ型のみを使用してください。

    • BLOB トピックの場合、MaxCompute テーブルスキーマには STRING 型の列が 1 つだけ含まれている必要があります。DataHub は自動的にこの列にデータを同期します。

      DataHub

      MaxCompute

      BLOB

      STRING

    • データ追跡とトラブルシューティングを容易にするには、MaxCompute テーブルの作成時に __rowkey__ STRING という名前の列を追加します。DataHub はこの列にトレース情報を自動的に同期することで、データ調査を支援します。

    アカウントと権限

    • MaxCompute にデータを同期するためのコネクタを作成する際には、MaxCompute アカウントの有効な認証情報を提供してください。ほとんどの場合、MaxCompute のサブアカウントで十分です。

    • このアカウントに、MaxCompute テーブルにアクセスするために必要な権限 (CreateInstance、Describe、Alter、Update) を付与してください。

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

    タイムスタンプ単位

    • TimestampUnit パラメーターは、TIMESTAMP データを宛先システムの DATETIME などの日付関連のデータ型に書き込む前に変換する単位を指定します。

    • TIMESTAMP 列に秒単位の値が含まれている場合は、コネクタ作成時に TimestampUnit に SECOND を選択してください。ミリ秒単位の値が含まれている場合は、MILLISECOND を選択してください。マイクロ秒単位の値が含まれている場合は、MICROSECOND を選択してください。

    重要

    MaxCompute の書き込み仕様により、パーティション数が多すぎると、DataHub からのデータ同期が遅くなります。MaxCompute の同期タスクを作成する場合、特に USER_DEFINE 同期モードでは、パーティションの数を可能な限り制限してください。

    • 同じパーティション内のデータをできるだけ連続させてください。パーティション間の頻繁な切り替えは避けてください。

    • 同期モードがパーティション作成を制御する場合、過剰な数のパーティションを作成しないでください。

    説明

    MaxCompute プロジェクトで 許可リスト 機能が有効になっている場合、許可リスト上のデバイスのみがプロジェクトにアクセスできます。MaxCompute の IP 許可リストを有効にした後、同期サービスがプロジェクトにアクセスできるように、サービスを許可リストに追加する必要があります。詳細については、「概要」をご参照ください。

    同期モード

    追加モード

    データはターゲットテーブルに追加されます。このモードは、データの追加のみが必要で更新は不要なシナリオに適しています。

    アップサートモード

    Upsert は、Update と Insert の組み合わせであり、次のロジックで動作します。

    • ターゲットテーブルに同じプライマリキーを持つレコードが存在する場合、既存のレコードが更新されます。

    • ターゲットテーブルに同じプライマリキーを持つレコードが存在しない場合、新しいレコードが挿入されます。

    アップサートモードは、データの更新と挿入を柔軟に処理する方法を提供し、ターゲットテーブルのデータを最新の状態に保ちます。

    説明

    MaxCompute のアップサート機能の詳細については、「基本概念」をご参照ください。

    ユースケース

    • プライマリキーに基づくデータ更新：データは時間とともに変化する可能性があり、既存のレコードをプライマリキーに基づいて更新する必要があります。

    • ターゲットテーブルでのデータの一意性の維持：このモードは、ターゲットテーブルでレコードの一意性を強制することにより、重複データを防ぎます。

    • 重複データの処理：特定のプライマリキーに基づいて大量のデータを重複排除します。

    設定

    1. DataHub トピックタイプ：TUPLE トピックである必要があります。

    2. DataHub トピックスキーマ：以下の 2 種類がサポートされています。

      1. DTS フォーマット：DTS が DataHub にデータを同期する際に使用されるスキーマタイプ。

      2. カスタムフォーマット：ユーザーが作成したスキーマです。操作列として機能する String 型の列を選択する必要があります。

    3. MaxCompute ターゲットテーブル：トランザクション 2.0 テーブルである必要があります。

    同期ルール

    DTS フォーマット

    DTS から DataHub に同期されるデータ形式の場合、DataHub はスキーマ内の固定カラム operation_flag、before_flag、および after_flag を使用して、以下のルールに基づいて ODPS ターゲットテーブルにデータを同期する方法を決定します：

    operation_flag

    before_flag

    after_flag

    操作タイプ

    アクション

    I

    *

    *

    UPSERT

    プライマリキーに基づいてターゲットテーブルのレコードを更新します。

    U

    Y

    N

    DELETE

    プライマリキーに基づいてターゲットテーブルからレコードを削除します。

    U

    N

    Y

    UPSERT

    プライマリキーに基づいてターゲットテーブルのレコードを更新します。

    D

    *

    *

    DELETE

    プライマリキーに基づいてターゲットテーブルからレコードを削除します。

    カスタムフォーマット

    ユーザーが作成したデータの場合、DataHub は選択された操作列を使用して、MaxCompute ターゲットテーブルにデータを同期する方法を決定します。

    操作値

    操作タイプ

    アクション

    U

    UPSERT

    プライマリキーに基づいてターゲットテーブルのレコードを更新します。

    D

    DELETE

    プライマリキーに基づいてターゲットテーブルからレコードを削除します。

    同期コネクタの作成

    1. DataHub コンソールで、トピックの詳細ページに移動します。

    2. トピック詳細ページの右上隅にある [同期] をクリックして、同期タスクを作成します。

    3. MaxCompute ジョブタイプを選択して、コネクタ作成ページを開きます。

      パラメーター：

      パラメーター

      オプション

      必須

      説明

      プロジェクト名

      /

      はい

      MaxCompute プロジェクトの名前。ドロップダウンリストからプロジェクトを選択できます。プロジェクトリストを取得する権限がない場合は、手動で名前を入力してください。

      スキーマ

      /

      いいえ

      MaxCompute スキーマの名前。

      説明

      スキーマ機能を使用するには、スキーマ構文開発を有効にする必要があります。これを有効にする方法とスキーマの詳細については、「スキーマ操作」をご参照ください。

      テーブル

      /

      はい

      MaxCompute テーブルの名前。ドロップダウンリストからテーブルを選択できます。テーブルリストを取得する権限がない場合は、手動で名前を入力してください。

      説明

      アップサートモードを使用する場合、ターゲットテーブルはトランザクション 2.0 テーブルである必要があります。

      同期モード

      追加

      はい

      ターゲットの MaxCompute テーブルにデータを追加します。

      アップサート

      プライマリキーに基づいて、ターゲットの MaxCompute テーブルのデータを更新または削除します。

      詳細については、同期モードのセクションをご参照ください。

      認証方法

      AK

      AccessKey ペアを使用して認証します。

      DataHub デフォルトロール

      このオプションを選択すると、datahub__access__role ロールがプロジェクトに自動的に承認されます。このロールの権限ポリシーは次のとおりです。

      {
  "Statement": [
    {
      "Action": [
        "odps:CreateInstance",
        "odps:CreateTable",
        "odps:Describe",
        "odps:Alter",
        "odps:Update"
      ],
      "Effect": "Allow",
      "Resource": [
        "*"
      ]
    }
  ],
  "Version": "1"
}

      カスタムロール

      RAM コンソールで作成および管理できるカスタムロール。

      アップサートメソッド

      SYNC_CUSTOM

      「同期モード」が「アップサート」の場合に必須です。「同期モード」が「追加」の場合は適用されません。

      カスタムのアップサート操作フィールドを使用します。

      SYNC_NONE

      すべてのデータは、アップサート操作を使用してターゲットテーブルに書き込まれます。

      SYNC_DTS

      データが DTS から DataHub に書き込まれ、新しい DTS 添付列ルールが有効になっているシナリオに適用されます。

      SYNC_DTS_OLD

      データが DTS から DataHub に書き込まれ、古い DTS 添付列ルールが有効になっているシナリオに適用されます。

      プライマリキーフィールド

      /

      ターゲットテーブルのプライマリキー列。

      アップサート操作フィールド

      /

      アップサートメソッドが SYNC_CUSTOM に設定されている場合は必須です。

      操作フィールドとして使用する String 型の列を選択します。このフィールドは、現在のデータがアップサート操作または削除操作としてダウンストリームテーブルに同期されるかどうかを示します。

    4. アップサートモードの詳細については、このトピックのアップサートモードのセクションをご参照ください。

      • インポートするフィールド：DataHub を設定して、列のサブセットのみを MaxCompute テーブルに同期させることができます。

      • パーティションモードは、データが書き込まれる MaxCompute パーティションを決定します。DataHub は、次のパーティション分割方法をサポートしています。

        パーティションモード

        パーティションの基準

        トピックタイプ

        説明

        USER_DEFINE

        レコード内のパーティション列の値。列名は MaxCompute のパーティションフィールドと同じ名前である必要があります。

        TUPLE

        • DataHub スキーマには、MaxCompute パーティションフィールドが含まれている必要があります。

        • 列の値は UTF-8 文字列 でなければなりません。値が空の場合、データがパーティション分割されていないことを示します。

        SYSTEM_TIME

        レコードが DataHub に書き込まれた時刻。

        TUPLE / BLOB

        • パーティション設定で、MaxCompute パーティションの時間変換形式を設定してください。

        • タイムゾーン情報を設定してください。

        EVENT_TIME

        レコードの event_time(TIMESTAMP) 列の値。

        TUPLE

        • パーティション設定で、MaxCompute パーティションの時間変換形式を設定してください。

        • タイムゾーン情報を設定してください。

        META_TIME

        レコードの __dh_meta_time__ 属性フィールドの値。

        TUPLE / BLOB

        • パーティション設定で、MaxCompute パーティションの時間変換形式を設定してください。

        • タイムゾーン情報を設定してください。

        SYSTEM_TIME、EVENT_TIME、および META_TIME の各モードでは、タイムスタンプとタイムゾーン設定に基づいてタイムスタンプが変換され、MaxCompute パーティションが作成されます。デフォルトの単位はマイクロ秒です。

      • パーティション設定は、タイムスタンプを MaxCompute パーティションに変換する方法を指定します。コンソールでは、MaxCompute パーティションにデフォルトの固定形式が使用されます。設定は次のとおりです。

        パーティション

        時間形式

        説明

        ds

        %Y%m%d

        日

        hh

        %H

        時

        mm

        %M

        分

        • パーティション間隔: タイムスタンプを MaxCompute パーティションに変換するための時間間隔です。範囲は 15 分から 1440 分 (1 日) で、増分は 15 分 です。

        • タイムゾーン情報 (TimeZone) は、タイムスタンプに基づいて MaxCompute パーティションを変換するために使用されるタイムゾーンを決定します。

        • 区切り文字：BLOB データを同期する場合、16 進数の区切り文字を指定して、MaxCompute に同期する前にデータを分割できます。 たとえば、0A は改行文字 (\n) を表します。

        • Base64 エンコーディング：DataHub はデフォルトで BLOB データをバイナリデータとして保存しますが、MaxCompute の対応する列は STRING 型です。そのため、コンソールで同期コネクタを作成すると、データはデフォルトで Base64 エンコードされます。より多くのカスタマイズオプションについては、SDK を使用してください。

    同期コネクタの表示

    コネクタの詳細ページに移動して、実行ステータスやチェックポイントなどの情報を表示できます。コネクタの再起動や停止などの操作も実行できます。

    同期コネクタの編集

    同期タスクページで [編集] をクリックして、認証方法、インポートされたフィールド、パーティション間隔、タイムゾーン、および TimestampUnit の値を変更できます。

    同期の例

    USER_DEFINE モード

    1. DataHub トピックを作成します。

      トピックスキーマには MaxCompute パーティションフィールドが含まれている必要があり、その型は STRING である必要があります。

    2. DataHub SDK を使用して DataHub トピックにデータを書き込みます。

      テストのために、SDK を使用して [ds,hh,mm] の値が [20210304,01,15] と [20210304,02,15] のレコードをいくつか書き込みます。

    3. 同期コネクタを作成します。

      USER_DEFINE パーティションモードでは、同期中にパーティション設定フィールドを設定できます。MaxCompute に対応するテーブルが存在しない場合、自動的に作成できます。インポートフィールド設定で、フィールド f1 と f2 を同期し、フィールド f3 を除外するように指定してください。

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

      DataHub コンソールでコネクタの同期情報を表示し、MaxCompute でデータ結果をクエリできます。

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

    SYSTEM_TIME モード

    1. DataHub トピックを作成します。

      パーティションは データが DataHub に書き込まれた時間 に基づいて計算されるため、トピックスキーマにはデータフィールドのみを含める必要があり、パーティションフィールドは必要ありません。

    2. DataHub SDK を使用して DataHub トピックにデータを書き込みます。

      テストのため、SDK を使用して複数のレコードを書き込みます。DataHub への現在の書き込み時刻は 2021-03-04 14:02:45 とします。

    3. 同期コネクタを作成します。

      パーティション設定が MaxCompute テーブルのパーティションと一致していることを確認してください。

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

      DataHub コンソールでコネクタの DoneTime などの同期情報を表示し、MaxCompute でデータ結果をクエリできます。

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

    よくある質問

    • MaxCompute のタイムスタンプフィールドが、同期後に 1970-01-19 のような日付で表示されるのはなぜですか？

      • 原因：DataHub から MaxCompute への同期のデフォルトの時間単位はマイクロ秒ですが、DataHub に書き込まれたタイムスタンプはミリ秒でした。

      • 解決策：タイムスタンプがマイクロ秒単位で DataHub に書き込まれるようにしてください。