embulk oracle

oracle-jdbc.rb

# -*- coding: utf-8 -*-
module Embulk
  class OutputOracle < OutputPlugin
    $CLASSPATH << "/home/kumagi/develop/embulk/ojdbc14.jar"
    require 'java'
    java_import 'oracle.jdbc.OracleDriver'
    java_import 'java.sql.DriverManager'

    Plugin.register_output('oracle', self)

    def self.transaction(config, schema, processor_count, &control)
      task = {
        'host' => config.param('host', :string, default: "127.0.0.1"),
        'port' => config.param('port', :integer, default: 1521),
        'username' => config.param('username', :string),
        'password' => config.param('password', :string),
        'tns' => config.param('tns', :string),
        'table' => config.param('table', :string),
      }

      puts "Oracle output started."
      commit_reports = yield(task)
      puts "Oracle output finished. commit_reports=#{commit_reports}"

      return {}
    end

    def initialize(task, schema, index)
      puts "Oracle output thread #{index}..."
      super

      puts "drivers{"
      DriverManager.getDrivers().each{|d|
        p d
      }
      puts "}"

      require 'java'

      java.lang.Class.forName("oracle.jdbc.driver.OracleDriver",
                              true,
                              java.lang.Thread.currentThread.getContextClassLoader)

      puts "drivers{"
      DriverManager.getDrivers().each{|d|
        p d
      }
      puts "}"

      @conn = DriverManager.get_connection(
         'jdbc:oracle:thin:@127.0.0.1:1521/XE',
          #"jdbc:oracle:thin:@#{task['host']}:#{task['port']}:#{task['tns']}",
          task['username'], task['password'])

      # drop if exists
      drop_stmt = @conn.create_statement
      dropquery = %q{DROP TABLE IF EXISTS #{task['table']};}
      drop_stmt.execute_query(dropquery)

      # create
      create_stmt = @conn.create_statement
      createquery = %q{CREATE TABLE #{task['table']};}
      create_stmt.execute_query(createquery)

      @rows = 0
    end

    def close
    end

    def add(page)
      prep = "INSERT INTO #{task['table']} VALUES (:1, :2)"
      begin
        insert = conn.prepare_statement(prep)
        page.each { |record|
          insert.set_int 1, 281
          insert.set_string 2, "test1"
          insert.execute
        }
      ensure
        prep.cloe
      end
    end

    def finish
    end

    def abort
    end

    def commit
      commit_report = {
        "rows" => @rows,
        "unique_keys" => @unique_keys.size
      }
      return commit_report
    end
  end

end