         job (filter s m)]
     (mapcat #(extract-text % (get source %)) job)))
 
-(mc/insert-batch "xuetest" (integrate-text :tianya "tianya" :douban 2 :tieba "tieba" :gada "haha" :weibo "weibo"))
+(insert-by-part "xuetest" (integrate-text :tianya "star_tianya_content"
+                                          :douban "star_douban_shortcomments"
+                                          :tieba "star_baidutieba_contents"
+                                          :gada "haha"
+                                          :weibo "star_weibo_history"
+                                          :youku "star_youku_video"))
+
 
 (defn extract-text
   [source-key source-address]
   (cond (= source-key :tianya) (extract-tianya source-address)
         (= source-key :tieba) (extract-tieba source-address)
         (= source-key :weibo) (extract-weibo source-address)
-        (= source-key :douban) (println "This is douban")
-        (= source-key :youku) (println "This is youku")))
+        (= source-key :douban) (extract-douban source-address)
+        (= source-key :youku) (extract-youku source-address)))
 
 (defn extract-tieba
   [source-address]

         (map f)
         (map g))))

;(extract-weibo "weibo")






(mc/insert-batch "xuetest" (extract-tieba "tieba"))

;(extract-tieba "star_baidutieba_contents")







  (:import [com.mongodb MongoOptions ServerAddress WriteConcern];the following two is for mongo use
           org.bson.types.ObjectId))

(mg/connect! {:host "192.168.3.53" :port 7017})

#_(mg/connect!)

(mg/set-db! (mg/get-db "test"))

#_(mg/set-db! (mg/get-db "star"))

(declare extract-text extract-tieba extract-tianya extract-weibo extract-douban extract-youku)

(defn insert-by-part
  [collection data]
  (let [parts (partition-all 500 data)]
    (map #(mc/insert-batch collection %) parts)))

(defn integrate-text
  [& {:as source}]
