adhoc.kt

import com.treasuredata.client.*
import com.google.common.base.Function
import org.msgpack.core.MessagePack
import org.msgpack.core.MessageUnpacker
import org.msgpack.value.ArrayValue
import org.msgpack.core.MessageFormat
import com.treasuredata.client.model.*
import java.io.InputStream
import java.io.File
import kotlin.String


fun fraction(array: ArrayValue ) {
  println(array)
}


fun main(args: Array<String>) {
  val client = TDClient.newClient();
  println("Start connecting to TreasureData Database.")
  val names:List<String> = client.listDatabases().map { db ->
    db.getName().toString()
  }.toList()
  names.map { name -> println("There is Database of ${name}.") }
  // prestoは3倍ぐらいhiveより早いが、メモリが全然足りない40G ~ 80Gで死んでしまう
  //val jobId = client.submit(TDJobRequest.newPrestoQuery("dac_aonesync",
  val jobId = client.submit(TDJobRequest.newHiveQuery("tech_batch",
      File("kotlinDriver.sql").readText() ));
  val backOff          = ExponentialBackOff()
  val job:TDJobSummary = client.jobStatus(jobId)
  while(!client.jobStatus(jobId).getStatus().isFinished()) {
    Thread.sleep(backOff.nextWaitTimeMillis().toLong())
  }
  val jobInfo:TDJob    = client.jobInfo(jobId)
  println("log:\n ${jobInfo.getCmdOut()}")
  println("error log:\n ${jobInfo.getStdErr()}")
  val unpacker = TDHandler().unpackerHandler(client, jobId)
  println("Unpackerが呼び出されました")
  while( unpacker.hasNext() ) {
    val array = unpacker.unpackValue().asArrayValue()
    fraction(array)
  }
  println("Finished access to TresureData Database.")
  System.exit(0)
}

Exception in thread "main" com.treasuredata.client.TDClientProcessingException: [EXECUTION_FAILURE] java.nio.channels.ClosedChannelException The root cause: java.nio.channels.ClosedChannelException
at com.treasuredata.client.TDHttpClient.submitRequest(TDHttpClient.java:410)
at com.treasuredata.client.TDHttpClient.call(TDHttpClient.java:609)
at com.treasuredata.client.TDHttpClient.call(TDHttpClient.java:575)
at com.treasuredata.client.TDClient.doGet(TDClient.java:193)
at com.treasuredata.client.TDClient.jobStatus(TDClient.java:594)
at TreasureDataKt.main(TreasureData.kt:31)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.jetbrains.kotlin.runner.AbstractRunner.run(runners.kt:60)
at org.jetbrains.kotlin.runner.Main.run(Main.kt:104)
at org.jetbrains.kotlin.runner.Main.main(Main.kt:110)