nodeからTreasureDataに接続する方法をメモ

コード

var TD = require('td');

var tdSelectData = (function() {
    var td_databases = 'test_db'; // DB名
    var td_api_key = '123/123456789987654321'; // APIキー
    var client = new TD(td_api_key); // TDクライアント
    var td_query = 'select * from test_table'; // 実行クエリ
    client.prestoQuery(td_databases, td_query, function(err, results) {

        console.log('=============TD results = ' + JSON.stringify(results) + '\r\n');

        const MAX_TIME = 600; 
        var time = 0;
        (function loop() {

            if (time < MAX_TIME) {

                client.showJob(results.job_id, function(err, r) {

                    console.log('TD JobStatus = ' + r.status + '\r\n');

                    if (r.status == 'error') {
                        console.log('TD Job error\r\n');
                        return false;
                    }

                    if (r && r.status == 'success') {
                        client.jobResult(results.job_id, 'csv', function(err, tdData) {
                            return tdData;
                        });
                    } else {
                        console.log('============this job is still processing...  waiting 5 seconds\r\n');
                        sl.sleep(5);
                        time += 5;
                        loop();
                    }
                });


            } else {
                console.log('timeout error\r\n');
                return false;
            }

        }());

    });
}());

解説

client.prestoQuery(td_databases, td_query, function(err, results){

prestoのjobを実行します。
callbackのresultsには以下のような値がJSON形式で返ってきます。

{ job: '123456789',
  database: 'test_db',
  job_id: '123456789' }

(function loop() {

こちらではTDのJOBの結果がSuccessになるまで、ループしています。
JOBの結果待ち時間は600秒としています。

client.showJob(results.job_id,function(err, r){

jobの結果を返します。
callbackのrには以下のような値がJSON形式で返ってきます。
実行中はstatusがrunning、完了後はsuccess、失敗はerrorとなります。

{ query: 'select * from test_table',
  type: 'presto',
  priority: 0,
  retry_limit: 0,
  duration: null,
  status: 'running',
  cpu_time: null,
  result_size: null,
  job_id: '123456789',
  created_at: '2017-12-12 02:22:03 UTC',
  updated_at: '2017-12-12 02:22:03 UTC',
  start_at: '2017-12-12 02:22:03 UTC',
  end_at: '',
  num_records: null,
  database: 'test_db',
  user_name: 'test_user',
  result: '',
  url: 'https://console.treasuredata.com/jobs/123456789',
  hive_result_schema: null,
  organization: null,
  debug: {省略}

client.jobResult(results.job_id, 'csv', function(err, tdData) {

jobの結果を出力します。
callbackのtdDataにはcsv形式でクエリの結果が出力されます。
各行ループする際は以下のようなコードを記載すればよいと思います。

client.jobResult(results.job_id, 'csv', function(err, tdData) {
   csv.parse(tdData, function(err, rawData) {
       console.log(rawData);
   });
});

参考