DataWorks:MaxCompute データソース

バッチ読み取り

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

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

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

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

• パーティション列を送信先テーブルに書き込むには、それらをカスタム列として追加します。

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

バッチ書き込み

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

• 送信先テーブルが DeltaTable の場合、[詳細設定] を展開し、[同期後の可視性] を [はい] に設定します。そうしないと、同時実行数が 1 より大きいシナリオでタスクが失敗します。

• MaxCompute 外部テーブルへのデータ書き込みはサポートされていません。

• 送信先列がソース列からマッピングされていない場合、同期後に NULL になり、テーブル作成時に指定されたデフォルト値を上書きします。

リアルタイム書き込み

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

• リアルタイム同期の送信先テーブルには、プライマリキーが必要です。

• MaxCompute 外部テーブルへのデータ書き込みはサポートされていません。

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

• ワンクリックリアルタイム同期タスクの場合、既存データは設定日にクエリできます。増分データは、翌日にマージされた後に利用可能になります。

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

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

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

• 現在のワークスペースと同じリージョンにあるカスタム MaxCompute データソースのみを使用できます。クロスリージョンのデータソースの場合、接続性テストは成功する可能性がありますが、MaxCompute でのテーブル作成フェーズ中に「engine not found」エラーで同期タスクが失敗します。

• MaxCompute がデータベース全体の同期の送信先として機能する場合、標準テーブルはリアルタイムデータベース全体の同期の増分ログモードとワンクリックリアルタイム完全増分同期のみをサポートします。対照的に、Delta Table はリアルタイムデータベース全体の同期とワンクリックリアルタイム完全増分同期の両方をサポートします。

  説明

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

データ型 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 の有効化

TIMESTAMP データ型を使用するには、MaxCompute データ型 2.0 を有効にする必要があります。Project Owner ロールを持つアカウントを使用して、次のコマンドを実行します。

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

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

MaxCompute コンピューティングリソースをワークスペースに関連付けると、デフォルトで DataWorks に 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"// 最大転送速度。単位は 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"// 最大転送速度。単位は 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 スクリプトのパラメーター