JDBCtap issue

db_tap.clj

(ns paduka.db_tap
  (:use cascalog.api)
  (:require [cascalog [vars :as v] [ops :as c] [workflow :as w]]
            [clojure.string :as s])
  (:import  [com.twitter.maple.jdbc JDBCScheme JDBCTap TableDesc]
            [cascading.tuple Fields])
  (:gen-class))

(defn mysql-tap []
  (let [scheme (JDBCScheme. (Fields. (into-array String ["id" "screen_name" "content"])) (into-array String ["id" "screen_name" "content"]))
        table-desc (TableDesc. "tweets")
        tap (JDBCTap. "jdbc:mysql://localhost:3306/burn?user=root&password=" "com.mysql.jdbc.Driver" table-desc scheme )]
    tap))

(def tweet-data
  [["mrjaba-1" "mrjaba" "tweet 1"]
  ["mrjaba-2" "mrjaba" "tweet 2"]])

(defn -main []
  (let [mysql (mysql-tap)]
    (?<- (mysql-tap) [?id] (tweet-data ?id _ _))))

(defn mysql-tap []
  (let [scheme (JDBCScheme. (Fields. (into-array String ["?id"])) (into-array String ["id"]))
        table-desc (TableDesc. "tweets" (into-array String ["?id"]) (into-array String ["id"]) (into-array String []))
        tap (JDBCTap. "jdbc:mysql://localhost:3306/burn?user=root&password=" "com.mysql.jdbc.Driver" table-desc scheme )]
    tap))