scala：StreamとLOAD DATA INFILEを使ったMySQL超高速データコピー(From DB2)

# ポイント
- 一気通貫。一時ファイルを作らない。
- select結果をStreamで取得→CSV形式に変換→load data local infileに直接食わす
- 再帰処理とcommit

##メイン
```main.scala

//再帰関数
def recursiveLoadFunction(s:Stream[UserTable]):Unit={
  //Streamの先頭から指定件数分を取得しListに変換。
  val xs=s.take(2000).toList;

  xs match{
    case z if z.length>0=>{ //リストが空でない場合
      //CSV(Stringデータ（カンマ、改行コード区切り））に変換する。
      //必要であれば項目追加やデータ加工を行う。
      val loadParams:String=z.map(x=>{
          List(x.userId,x.x.name,x.address)
          .mkString(",")
        }).mkString("\n")
      //load
      MysqlDb.load(loadParams)
      //残りを再帰処理
      recursiveLoadFunction(s.drop(len))
    }
  }
} 

//caseクラス（例）
case class UserTable(
 userId:Long,
 name:String,
 address:String
)

object MainObject{
  def main(args: Array[String]){
    //select結果をStreamで取得
    val s:Stream[UserTable]=Db2Db.getDataStream
    //実行
    recursiveLoadFunction(s)
    //MySql 切断
    MysqlDb.close
  }
}
```

##mysql
```mysql.scala
import scalikejdbc._
import scalikejdbc._, SQLInterpolation._
import com.mysql.jdbc.Driver
import com.ibm.icu.text.Transliterator
import org.joda.time._
import java.util.Date
import scala.collection.immutable.Stream;
import java.sql.{Connection,DriverManager}
import java.io.{BufferedReader, ByteArrayInputStream, ByteArrayOutputStream, InputStreamReader}

object MysqlDb{
  val url = "jdbc:mysql://【ホスト名】:【ポート番号】/【DB名】"
  val driver = "com.mysql.jdbc.Driver"
  val username = "【ユーザー名】"
  val password = "【パスワード】"
  var connection:Connection = _

  try {
    Class.forName(driver)
    this.connection = DriverManager.getConnection(url, username, password)
    this.connection.setAutoCommit(false);
  } catch {
    case e: Exception => e.printStackTrace
  }

  def load(loadParams:String)={
    try {
      var pstmt = this.connection.prepareStatement("""load data local infile '' """
          + """into table 【DB名】.【テーブル名】 fields terminated by ','""");
      val bais = new ByteArrayInputStream(loadParams.getBytes())
      pstmt.asInstanceOf[com.mysql.jdbc.Statement].setLocalInfileInputStream(bais)
      pstmt.execute();
      this.connection.commit();
    } catch {
      case e: Exception => e.printStackTrace
    }
  }

  def close={
    try {
      this.connection.close
    } catch {
      case e: Exception => e.printStackTrace
    }
  }
```

##db2 
```db2.scala
import scalikejdbc._
import scalikejdbc._, SQLInterpolation._
import com.ibm.icu.text.Transliterator
import org.joda.time._
import java.util.Date
import scala.collection.immutable.Stream

object Db2Db{
  Class.forName("com.ibm.db2.jcc.DB2Driver");
  ConnectionPool.singleton("jdbc:db2://【ホスト名】:【ポート番号】/【DB名】", "【ユーザー名】", "【パスワード】")
  GlobalSettings.loggingSQLAndTime = LoggingSQLAndTimeSettings(
    enabled = true,
    singleLineMode = true,
    logLevel = 'DEBUG
  )

  def getDataStream={
     val x:Stream[UserTable]= DB readOnly { implicit session =>
      sql"""
        SELECT USER_ID,NAME,ADDRESS FROM USER_TABLE 
      """
        .fetchSize(2000)
        .map(rs=>
            UserTable(
              rs.long("USER_ID"),
              rs.string("NAME"),
              rs.string("ADDRESS")
            )
        ).list.apply() toStream //Streamに変換
      }
     x
   }
}
```