DataWorks:MaxCompute

バッチ読み取り

• MaxCompute Reader は、パーティションテーブルと非パーティションテーブルからのデータ読み取りをサポートしますが、仮想ビューや外部テーブルからの読み取りはサポートしません。

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

• このメソッドを使用して、スケジューリング時間に対応するパーティションからデータを同期します。

  たとえば、t0 という名前のパーティションテーブルに id 列と name 列が含まれているとします。レベル 1 のパーティションキーは pt で、レベル 2 のパーティションキーは ds です。pt=<業務日> および ds=hangzhou のパーティションからデータを読み取るには、ソースを設定する際にパーティション値を pt=${scheduling parameter} および ds=hangzhou として指定する必要があります。その後、id 列と name 列のフィールドマッピングを設定できます。

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

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

バッチ書き込み

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

• 送信先テーブルが DeltaTable の場合は、Advanced Configuration を展開し、Visible After Synchronization を Yes に設定します。そうしないと、同時実行数が 1 より大きい場合にタスクがエラーを報告します。

• ソースから MaxCompute 外部テーブルへのデータ同期はサポートされていません。

• 送信先テーブルの列がソース列にマッピングされていない場合、テーブル作成時にデフォルト値が指定されていても、Data Integration は同期後にその値を null に設定します。

リアルタイム書き込み

• リアルタイムデータ同期タスクは、サーバーレスリソースグループをサポートします。

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

• ソースから MaxCompute 外部テーブルへのデータ同期はサポートされていません。

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

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

• MaxCompute へのワンクリックリアルタイム同期タスクは、毎日完全データパーティションを生成します。過剰なストレージ使用量を防ぐため、自動的に作成される MaxCompute テーブルにはデフォルトで 30 日間のライフサイクルが設定されています。この期間が不適切な場合は、タスク設定時に MaxCompute テーブル名をクリックしてライフサイクルを変更できます。

• Data Integration は、データのアップロードとダウンロードに MaxCompute エンジンのデータ同期チャネルを使用します。これらのチャネルのサービスレベルアグリーメント (SLA) については、「データ転送サービス (アップロード) のシナリオとツール」をご参照ください。これらのチャネルの SLA に基づいて、データ同期メソッドを評価してください。

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

• Data Integration は、現在のワークスペースと同じリージョンにあるユーザー作成の MaxCompute データソースのみをサポートします。異なるリージョンのデータソースの接続テストが成功しても、テーブル作成中に同期タスクが失敗し、「engine does not exist」エラーが報告されます。

• MaxCompute がデータベース全体の同期の送信先である場合、通常のテーブルに対してはデータベース全体の完全および増分タスクとリアルタイム増分ストリームモードのみをサポートします。送信先テーブルが DeltaTable の場合、リアルタイム同期とデータベース全体の完全および増分タスクの両方がサポートされます。

  説明

  ユーザー作成の MaxCompute データソースを使用する場合でも、MaxCompute エンジンを DataWorks プロジェクトに関連付ける必要があります。そうしないと、MaxCompute SQL ノードを作成できず、完全同期のための 'done' マーカーノードが作成されなくなります。

MaxCompute 1.0 データ型

MaxCompute 2.0 および Hive 互換データ型

データ型変換

次の表に、MaxCompute Reader のデータ型変換ルールを示します。

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

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

• MaxCompute で必要なプロジェクトレベルの設定を有効にします。これらの操作には、プロジェクトオーナーの権限が必要です。MaxCompute の権限の詳細については、「ロール計画」をご参照ください。

ACID セマンティクスの有効化

ACID セマンティクスを有効にするには、プロジェクトオーナーとしてクライアントで次のコマンドを実行します。MaxCompute の ACID セマンティクスの詳細については、「ACID セマンティクス」をご参照ください。

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

(オプション) V2.0 データ型の有効化

TIMESTAMP データ型を使用するには、プロジェクトオーナーとしてクライアントで次のコマンドを実行して、MaxCompute V2.0 データ型を有効にします。

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

(オプション) アカウントの権限付与

MaxCompute コンピュートエンジンをワークスペースにバインドすると、DataWorks に MaxCompute データソースが自動的に作成されます。このデータソースを使用して、現在のワークスペース内でデータ同期を行うことができます。別のワークスペースでこの MaxCompute データソースからデータを同期するには、別のワークスペースのデータソースのアクセスアカウントが、この MaxCompute コンピュートエンジンにアクセスするために必要な権限を持っていることを確認してください。クロスアカウント認証の詳細については、「クロスアカウント認証 (MaxCompute および Hologres)」をご参照ください。

単一テーブルのバッチ同期タスク

• 「コードレス UI でのバッチ同期の設定」および「コードエディタの使用」をご参照ください。

• コードエディタのパラメーターの完全なリストとサンプルスクリプトについては、「付録：サンプルスクリプトとパラメーター」をご参照ください。

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

「単一テーブルのリアルタイム同期タスクの設定」をご参照ください。

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

「データベース全体のバッチ同期タスクの設定」、「データベース全体のリアルタイム同期タスクの設定」、および「データベース全体の完全および増分同期タスクの設定」をご参照ください。

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

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

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,// 速度制限を有効にするかどうかを指定します。このパラメーターが true に設定されている場合、mbps パラメーターが有効になります。
            "concurrent":1, // ジョブの同時実行数。
            "mbps":"12"// 速度制限のレート。1 mbps = 1 MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

MaxCompute のトンネルエンドポイントを指定するには、スクリプトモードでデータソースを手動で設定できます。上記のサンプルの "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,// クリーンアップルール。
                "isCompress":false,// データを圧縮するかどうかを指定します。
                "datasource":"odps_first",// データソース名。
            "column": [// 送信先列名。
                "id",
                "name",
                "age",
                "sex",
                "salary",
                "interest"
                ],
                "table":""// 送信先テーブル名。
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// エラーレコード数。許可されるダーティデータの最大レコード数です。
        },
        "speed":{
            "throttle":true,// 速度制限を有効にするかどうかを指定します。このパラメーターが true に設定されている場合、mbps パラメーターが有効になります。
            "concurrent":1, // ジョブの同時実行数。
            "mbps":"12"// 速度制限のレート。1 mbps = 1 MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

MaxCompute のトンネルエンドポイントを指定するには、スクリプトモードでデータソースを手動で設定できます。上記のサンプルの "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 スクリプトパラメーター