大量のデータをエクスポートするためのソリューション - MaxCompute

このトピックでは、MaxCompute SQL 計算の結果をエクスポートするためのいくつかのメソッドについて説明します。

説明

このトピックの SDK の例は Java で提供されています。

概要

SQL 文の結果をエクスポートするには、次のメソッドを使用できます。

少量のデータの場合は、SQL タスクを使用してすべてのクエリ結果を取得できます。
テーブル全体またはパーティションをエクスポートするには、Tunnel を使用してクエリ結果をエクスポートできます。
SQL 文が複雑な場合は、Tunnel と SQL を組み合わせてクエリ結果をエクスポートできます。
DataWorks を使用して SQL 文を実行し、データを同期し、時間指定スケジューリングや依存関係設定などの機能を使用できます。
オープンソースツール DataX を使用すると、MaxCompute から宛先データソースにデータを簡単にエクスポートできます。

SQLTask を使用したデータのエクスポート

SDK が提供する SQLTask を使用して SQLTask.getResult(i) インターフェイスを呼び出すと、SQL 文を実行してその結果を取得できます。詳細については、「SQLTask」をご参照ください。

SQLTask を使用する場合は、次の点に注意してください。

SQLTask.getResult(i) は SELECT クエリの結果をエクスポートするために使用されます。 show tables などの他の MaxCompute コマンド操作の結果をエクスポートするために使用することはできません。
SELECT 文がクライアントに返すデータレコードの数は、READ_TABLE_MAX_ROW を使用して設定できます。詳細については、「プロジェクト操作」をご参照ください。
SELECT 文は、最大 10,000 件のデータレコードをクライアントに返すことができます。 SQLTask の使用を含め、クライアントで SELECT 文を直接実行すると、SELECT 文の末尾に LIMIT N を追加するのと同じことになります。

Tunnel を使用したデータのエクスポート

エクスポートする必要があるクエリ結果が、テーブル全体または特定のパーティションのコンテンツである場合は、Tunnel を使用できます。詳細については、「コマンドラインインターフェイス」および「Tunnel SDK」をご参照ください。

次のコードは、Tunnel コマンドラインを使用してデータをエクスポートする簡単な例を示しています。 Tunnel コマンドラインでサポートされていないシナリオの場合は、Tunnel 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

SQLTask と Tunnel を使用したデータのエクスポート

SQLTask は 10,000 件を超えるデータレコードを処理できませんが、Tunnel は処理できます。これらは相互に補完するため、組み合わせて使用することで 10,000 件を超えるデータレコードをエクスポートできます。

次のコードに例を示します。

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 データ同期を使用したデータのエクスポート

説明

この例では、DataWorks の基本モードを使用します。ワークスペースを作成するときは、[DataStudio のパブリックプレビューに参加] オプションを選択しないでください。この例は、パブリックプレビューのワークスペースには適用されません。

DataWorks を使用すると、SQL 文を実行し、データ同期タスクを構成して、データ生成とエクスポートの要件を満たすことができます。

DataWorks コンソールにログインします。
左側のナビゲーションウィンドウで、[ワークスペース管理] をクリックします。
目的のワークスペースの [アクション] 列で、[クイックアクセス] > [データ開発] を選択します。
ビジネスプロセスを作成します。
1. [ビジネスプロセス] を右クリックし、[新しいビジネスプロセス] を選択します。
2. [ビジネス名] を入力します。
3. [作成] をクリックします。
SQL ノードを作成します。
1. ビジネスプロセスを右クリックし、[新しいノード] > [MaxCompute] > [ODPS SQL] を選択します。
2. [ノード名] を runsql に設定し、[確認] をクリックします。
3. ODPS SQL ノードを構成し、[保存] をクリックします。
データ同期ノードを作成します。
1. ビジネスプロセスを右クリックし、[新しいノード] > [データ統合] > [オフライン同期] を選択します。
2. [ノード名] を sync2mysql に設定し、[確認] をクリックします。
3. データソースと宛先を選択します。
4. フィールドマッピングを構成します。
5. チャネルコントロールを構成します。
6. [保存] をクリックします。
データ同期ノードを ODPS SQL ノードに接続して、依存関係を構成します。このワークフローでは、ODPS SQL ノードがデータを生成し、データ同期ノードがデータをエクスポートします。

ワークフローのスケジューリングを構成した後 (デフォルトの構成を使用できます)、[実行] をクリックします。以下にデータ同期の操作ログを示します。

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

次の SQL 文を実行して、データ同期の結果を表示します。
```
select count(*) from result_in_db;
```