DataWorks:MaxCompute データソース

オフライン読み取り

• MaxCompute Reader は、パーティションテーブルと標準テーブルからの読み取りをサポートしています。仮想ビューからの読み取りや外部テーブルの同期はサポートしていません。

• MaxCompute のパーティションテーブルからオフラインで読み取りを行う場合、パーティションフィールドを直接マッピングすることはできません。パーティションフィールドの値を同期するには、カスタムフィールドを追加し、パーティション名を手動で入力してから、フィールドマッピングを実行する必要があります。

• スケジューリングパラメーターを使用してパーティションを指定し、自動的に置き換えることができます。これは、スケジューリング時間に基づいてパーティションを同期する必要があるシナリオで役立ちます。

  たとえば、t0 という名前のパーティションテーブルに id と name というフィールドがあるとします。ハッシュパーティションは pt、サブパーティションは ds です。pt=<データタイムスタンプ> かつ ds=hangzhou のパーティションからデータを読み取るには、データソース設定でパーティションの値を pt=${scheduling parameter} および ds=hangzhou として指定する必要があります。その後、フィールドマッピング設定時に id と name フィールドをマッピングできます。

• パーティションフィールドをカスタムフィールドとして追加することで、ターゲットテーブルに書き込むことができます。

• MaxCompute Reader は、WHERE 句を使用したデータフィルタリングをサポートしています。

オフライン書き込み

• MaxCompute Writer は、データに null 値が含まれている場合、VARCHAR 型をサポートしません。

• DeltaTable にデータを書き込む場合は、[同期後の可視性] を [はい] に設定します。そうしないと、同時実行数が 1 より大きい場合にタスクが失敗します。

  

リアルタイム書き込み

• リアルタイムデータ同期タスクは、サーバーレスリソースグループ (推奨) とデータ統合専用リソースグループをサポートしています。

• リアルタイムデータ同期タスクは、プライマリキーのないテーブルをサポートしていません。

• デフォルトの MaxCompute データソース (通常は odps_first) にリアルタイムで同期する場合、デフォルトで一時的な AccessKey (AK) が使用されます。一時的な AK は 7 日後に有効期限が切れるため、タスクは失敗します。プラットフォームは、一時的な AK の有効期限切れによる失敗を検出すると、タスクを自動的に再起動します。この種のエラーに対してモニタリングアラートを設定している場合は、アラートが届きます。

• MaxCompute へのワンクリックリアルタイム同期タスクでは、設定した日にのみ完全な履歴データをクエリできます。増分データは、翌日にマージが完了した後にのみ MaxCompute でクエリできます。

• MaxCompute へのワンクリックリアルタイム同期タスクは、毎日完全なパーティションを生成します。過剰なデータがストレージリソースを消費するのを防ぐため、このソリューションによって自動的に作成される MaxCompute テーブルのデフォルトのライフサイクルは 30 日です。この期間がビジネス要件を満たさない場合は、タスク設定中に MaxCompute テーブル名をクリックしてライフサイクルを変更できます。

• データ統合は、MaxCompute エンジンのデータチャネルを使用してデータのアップロードとダウンロードを行います。データチャネルの Service-Level Agreement (SLA) の詳細については、「Data Transmission Service (アップロード) のシナリオとツール」をご参照ください。MaxCompute エンジンのデータチャネルの SLA に基づいてデータ同期ソリューションを評価する必要があります。

• インスタンスモードで MaxCompute へのワンクリックリアルタイム同期を行う場合、データ統合専用リソースグループには少なくとも 8 vCPU と 16 GB のメモリが必要です。

• 現在のワークスペースと同じリージョンにあるユーザー作成の MaxCompute データソースのみがサポートされます。データソースの接続をテストする際には、リージョンをまたいだ MaxCompute プロジェクトに正常に接続できます。ただし、同期タスクが実行されると、MaxCompute でのテーブル作成フェーズで「engine not found」エラーが発生します。

• MaxCompute がデータベース全体の同期先である場合、サポートされる同期方法はテーブルのタイプによって異なります。標準テーブルの場合、データベース全体の完全同期と増分同期 (ほぼリアルタイム) のみがサポートされます。Delta Table の場合、リアルタイム同期とデータベース全体の完全同期と増分同期 (ほぼリアルタイム) の両方がサポートされます。

  説明

  ユーザー作成の MaxCompute データソースを使用する場合でも、DataWorks プロジェクトは MaxCompute エンジンにアタッチされている必要があります。そうしないと、MaxCompute SQL ノードを作成できず、完全同期のための「done」ノードの作成が失敗します。

データ型 1.0 でサポートされるフィールド

データ型 2.0 および Hive 互換データ型でサポートされるフィールド

データ型のマッピング

次の表に、MaxCompute Reader のデータ型のマッピングを示します。

MaxCompute への接続とプロジェクトレベル設定の有効化

• MaxCompute クライアントにログインします。詳細については、「ローカルクライアント (odpscmd) を使用した接続」をご参照ください。

• MaxCompute のプロジェクトレベルの設定を有効にします。必要な権限があることを確認してください。Project Owner アカウントを使用してこれらの操作を実行できます。MaxCompute の権限の詳細については、「ロール計画」をご参照ください。

ACID 属性の有効化

Project Owner アカウントを使用してクライアントで次のコマンドを実行し、ACID 属性を有効にすることができます。MaxCompute の ACID セマンティクスの詳細については、「ACID セマンティクス」をご参照ください。

setproject odps.sql.acid.table.enable=true;

(任意) データ型 2.0 の有効化

MaxCompute データ型 2.0 の timestamp 型を使用するには、Project Owner アカウントを使用してクライアントで次のコマンドを実行し、データ型 2.0 を有効にします。

setproject odps.sql.type.system.odps2=true;

(任意) 権限の付与

MaxCompute 計算リソースをワークスペースにアタッチすると、デフォルトで DataWorks に MaxCompute データソースが作成されます。このデータソースを使用して、現在のワークスペース内でデータを同期できます。別のワークスペースでこの MaxCompute データソースからデータを同期するには、別のワークスペースのデータソースに指定されたアクセスアカウントが、この MaxCompute プロジェクトへのアクセス権限を持っていることを確認する必要があります。クロスアカウント認証の詳細については、「クロスアカウント認証 (MaxCompute および Hologres)」をご参照ください。

単一テーブルのオフライン同期タスクの設定

• 手順については、「コードレス UI でのタスクの設定」および「コードエディタでのタスクの設定」をご参照ください。

• コードエディタでの設定に関するすべてのパラメーターとスクリプトデモについては、「付録：スクリプトデモとパラメーターの説明」をご参照ください。

単一テーブルのリアルタイム同期タスクの設定

手順については、「データ統合でのリアルタイム同期タスクの設定」および「DataStudio でのリアルタイム同期タスクの設定」をご参照ください。

データベース全体の同期タスクの設定

手順については、「データベース全体のオフライン同期」、「データベース全体のリアルタイム同期」、および「データベース全体の完全同期と増分同期 (ほぼリアルタイム)」をご参照ください。

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

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

Reader スクリプトデモ

{
    "type":"job",
    "version":"2.0",
    "steps":[
        {
            "stepType":"odps",// プラグイン名。
            "parameter":{
                "partition":[],// データを読み取るパーティション。
                "isCompress":false,// データを圧縮するかどうかを指定します。
                "datasource":"",// データソース。
                "column":[// ソーステーブルの列。
                    "id"
                ],
                "where": "",// WHERE 句を使用してデータをフィルタリングする場合は、具体的な句を入力します。
                "enableWhere":false,// WHERE 句を使用してデータをフィルタリングするかどうかを指定します。
                "table":""// テーブル名。
            },
            "name":"Reader",
            "category":"reader"
        },
        { 
            "stepType":"stream",
            "parameter":{
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// エラーレコード数。
        },
        "speed":{
            "throttle":true,// throttle が false に設定されている場合、mbps パラメーターは有効にならず、データレートは制限されません。throttle が true に設定されている場合、データレートは制限されます。
            "concurrent":1, // 同時実行ジョブ数。
            "mbps":"12"// データレートの制限。1 mbps = 1 MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

MaxCompute の Tunnel エンドポイントを指定するには、コードエディタでデータソースを手動で設定できます。上記の例の "datasource":"", を特定のデータソースパラメーターに置き換えます。次のコードは例です。

"accessId":"*******************",
"accessKey":"*******************",
"endpoint":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api",
"odpsServer":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api", 
"tunnelServer":"http://dt.eu-central-1.maxcompute.aliyun.com", 
"project":"*****", 

Reader スクリプトのパラメーター

Writer スクリプトデモ

次のコードは、サンプルスクリプト設定です。

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"odps",// プラグイン名。
            "parameter":{
                "partition":"",// パーティション情報。
                "truncate":true,// クリーンアップルール。
                "compress":false,// データを圧縮するかどうかを指定します。
                "datasource":"odps_first",// データソース名。
            "column": [// ソース列名。
                "id",
                "name",
                "age",
                "sex",
                "salary",
                "interest"
                ],
                "table":""// テーブル名。
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// エラーレコード数。これは許容されるダーティデータの最大レコード数です。
        },
        "speed":{
            "throttle":true,// throttle が false に設定されている場合、mbps パラメーターは有効にならず、データレートは制限されません。throttle が true に設定されている場合、データレートは制限されます。
            "concurrent":1, // 同時実行ジョブ数。
            "mbps":"12"// データレートの制限。1 mbps = 1 MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

MaxCompute の Tunnel エンドポイントを指定するには、コードエディタでデータソースを手動で設定できます。上記の例の "datasource":"", を特定のデータソースパラメーターに置き換えます。次のコードは例です。

"accessId":"<yourAccessKeyId>",
 "accessKey":"<yourAccessKeySecret>",
 "endpoint":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api",
 "odpsServer":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api", 
"tunnelServer":"http://dt.eu-central-1.maxcompute.aliyun.com", 
"project":"**********", 

Writer スクリプトのパラメーター