DataWorks:MaxCompute データソース

バッチ読み取り

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

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

• スケジューリングパラメーターを使用してパーティション値を指定し、自動置換を有効にすることで、スケジューリング時間に基づいて対応するパーティションからデータが同期されるようにできます。

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

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

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

バッチ書き込み

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

• 宛先テーブルが DeltaTable の場合は、詳細設定 を展開し、同期完了後、表示されます。 を はい に設定します。そうしないと、同時実行数が 1 より大きい場合にタスクでエラーが報告されます。

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

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

• MaxCompute Writer は、インポート前 SQL 文 (preSql) やインポート後 SQL 文 (postSql) をサポートしません。データの書き込み前後に宛先テーブルで SQL 操作を実行する必要がある場合 (たとえば、増分書き込みのために部分的なデータを削除する DELETE 文を実行するなど)、データ同期タスクの上流または下流に SQL ノードを設定して、必要な前処理または後処理操作を処理します。

リアルタイム書き込み

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

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

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

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

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

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

• Data Integration は、MaxCompute エンジンの同期データチャネルを使用してデータをアップロードおよびダウンロードします。同期データチャネルの SLA の詳細については、「MaxCompute Tunnel の概要」をご参照ください。MaxCompute エンジンの同期データチャネルの SLA に基づいて、データ同期技術の選択を評価してください。

• インスタンスモードで MaxCompute へのワンクリックリアルタイム同期を行う場合、Data Integration 専用リソースグループには最低 8C16G の仕様が必要です。

• 現在のワークスペースと同じリージョンにあるカスタム MaxCompute データソースのみがサポートされます。リージョン間の MaxCompute プロジェクトは接続性テストに合格する可能性がありますが、タスク実行中に MaxCompute でのテーブル作成フェーズでエンジンが存在しないことを示すエラーが報告されます。

• データベース全体の同期の宛先として MaxCompute を使用する場合、テーブルタイプが通常のテーブルであれば、MaxCompute へのワンクリックリアルタイム同期と、データベース全体のリアルタイム同期の増分ストリーミングモードのみがサポートされます。テーブルタイプが Delta Table の場合、データベース全体のリアルタイム同期とMaxCompute へのワンクリックリアルタイム同期の両方がサポートされます。

  説明

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

1.0 データ型でサポートされる列

2.0 データ型および Hive 互換データ型でサポートされる列

データ型の変換

次の表に、MaxCompute Reader がサポートするデータ型の変換を示します。

データ変換が失敗した場合、またはデータが宛先データソースに書き込まれなかった場合、そのデータはダーティデータとして扱われます。これはダーティデータのしきい値と組み合わせて使用できます。

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

• MaxCompute クライアントにログインします。詳細については、「MaxCompute クライアント」をご参照ください。

• MaxCompute プロジェクトレベル設定の有効化：必要な権限があることを確認してください。Project Owner アカウントを使用して関連操作を実行できます。MaxCompute の権限の詳細については、「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 へのワンクリックリアルタイム同期」をご参照ください。

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

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

Reader スクリプトのデモ

実行する前に、次のコードからコメントを削除してください。

{
    "type":"job",
    "version":"2.0",
    "steps":[
        {
            "stepType":"odps",//プラグイン名。
            "parameter":{
                "partition":[],//データが読み取られるパーティション。
                "isCompress":false,//データを圧縮するかどうかを指定します。
                "datasource":"",//データソース。
                "column":[//ソーステーブルの列情報。
                    "id"
                ],
                "where": "",//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 の トンネルエンドポイントを指定したい場合は、スクリプトモードでデータソースを手動で設定できます。上記の例の "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,//throttle が false に設定されている場合、mbps パラメーターは効果がなく、速度制限は無効になります。throttle が true に設定されている場合、速度制限は有効になります。
            "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 スクリプトのパラメーター