MaxCompute:一括データエクスポートソリューション

SQL クエリの結果をエクスポートするには、以下の方法を利用できます。

• 小規模データの場合、SQLTask を使用してクエリ結果全体を取得します。

• テーブル全体または特定のパーティションをエクスポートする場合は、Tunnel を使用します。

• 複雑な SQL クエリの場合は、SQLTask と Tunnel を組み合わせて使用します。

• DataWorks を使用して SQL クエリを実行し、データ同期を実施し、定期タスクや依存関係を設定します。

• オープンソースツール DataX を使用して、MaxCompute から送信先データソースへデータをエクスポートします。

注意事項：

• SQLTask.getResult(i) は SELECT クエリの結果のみをエクスポートします。show tables などの他のコマンドはサポートしていません。

• READ_TABLE_MAX_ROW プロパティは、SELECT 文が返すレコード数を制御します。このプロパティは プロジェクト操作 で設定します。

• SELECT 文はクライアントに対して最大 10,000 レコードを返します。クライアント上で SELECT 文を実行する（SQLTask 経由を含む）ことは、LIMIT N を末尾に追加することと同等です。

以下の例では、Tunnel コマンドを使用してデータをエクスポートしています。CLI でサポートされていない SDK ベースのシナリオについては、「バッチデータチャネルの概要」をご参照ください。

tunnel d wc_out c:\wc_out.dat;
2016-12-16 19:32:08 - new session: 201612161932082d3c9b0a012f68e7 total lines: 3
2016-12-16 19:32:08 - file [0]: [0, 3), c:\wc_out.dat
downloading 3 records into 1 file
2016-12-16 19:32:08 - file [0] start
2016-12-16 19:32:08 - file [0] OK. total: 21 bytes
download OK

例：

Odps odps = OdpsUtils.newDefaultOdps(); // Odps オブジェクトを初期化します。
Instance i = SQLTask.run(odps, "select * from wc_in;");
i.waitForSuccess();
// InstanceTunnel を作成します。
InstanceTunnel tunnel = new InstanceTunnel(odps);
// インスタンス ID を基に DownloadSession を作成します。
InstanceTunnel.DownloadSession session = tunnel.createDownloadSession(odps.getDefaultProject(), i.getId());
long count = session.getRecordCount();
// レコード数を出力します。
System.out.println(count);
// データ取得の方法は TableTunnel と同じです。
TunnelRecordReader reader = session.openRecordReader(0, count);
Record record;
while((record = reader.read()) != null)
{
    for(int col = 0; col < session.getSchema().getColumns().size(); ++col)
    {
        // wc_in テーブルのカラムはすべて STRING 型です。直接出力するか、ローカルファイルに書き込むことができます。
        System.out.println(record.get(col));
    }
}
reader.close();

DataWorks を使用して SQL クエリを実行し、データ同期タスクを設定します。

1. DataWorks コンソールにログインします。

2. 左側のナビゲーションウィンドウで、[ワークスペース] をクリックします。

3. 対象のワークスペースの [操作] 列で、[ショートカット] > データ開発 を選択します。

4. ワークフローを作成します。

   1. [ビジネスフロー] を右クリックし、[ワークフローの作成] を選択します。

   2. [名前] を入力します。

   3. [作成] をクリックします。

5. SQL ノードを作成します。

   1. ワークフローを右クリックし、[ノードの作成] > [MaxCompute] > [ODPS SQL] を選択します。

   2. [名前] に runsql を設定し、[OK] をクリックします。

   3. ODPS SQL ノードを設定し、[保存] をクリックします。

6. データ同期ノードを作成します。

   1. ワークフローを右クリックし、[ノードの作成] > [データ統合] > [オフライン同期] を選択します。

   2. [名前] に sync2mysql を設定し、[OK] をクリックします。

   3. データソースと送信先を選択します。

   4. フィールドマッピングを設定します。

   5. チャンネルコントロールを設定します。

   6. [保存] をクリックします。

7. ODPS SQL ノードをデータ同期ノードに接続し、SQL ノードが同期ノードがエクスポートするデータを生成するようにします。

8. ワークフロースケジューリングを設定するかデフォルトを使用し、[実行] をクリックします。実行ログの例：

   2016-12-17 23:43:46.394 [job-15598025] INFO JobContainer - 
   Task start time : 2016-12-17 23:43:34
   Task end time : 2016-12-17 23:43:46
   Total time consumed : 11s
   Average traffic : 31.36KB/s
   Record write speed : 1668rec/s
   Total records read : 16689
   Total read/write failures : 0

9. 同期結果を確認します：

   select count(*) from result_in_db;