mysql表导出到mongodb_一个将mysql表导入到mongodb的ruby脚本

功能&＃xff1a;

将mysql指定表的数据导入到mongodb的指定表&＃xff0c;导入过程保证数据不丢失&＃xff0c;如果数据有更新也会重新再导。

要求&＃xff1a;

源表必须要有两个字段&＃xff1a;id:主键 mmm_ts:最后更新时间戳

require &＃39;rubygems&＃39;

require &＃39;mongo&＃39;

require &＃39;active_record&＃39;

mongo_server &＃61;

{:local &＃61;>

{:server &＃61;> "localhost", :port &＃61;> 27017}

}

mysql_server &＃61;

{:local &＃61;>

{

:adapter &＃61;> &＃39;mysql&＃39;,

:host &＃61;> &＃39;localhost&＃39;,

:username &＃61;> &＃39;root&＃39;,

:password &＃61;> &＃39;$$$&＃39;,

:encoding &＃61;> &＃39;utf8&＃39;

} ,

:main_db &＃61;>

{

:adapter &＃61;> &＃39;mysql&＃39;,

:host &＃61;> &＃39;mydb.com&＃39;,

:username &＃61;> &＃39;abc&＃39;,

:password &＃61;> &＃39;cba&＃39;,

:encoding &＃61;> &＃39;utf8&＃39;

}

}

#从mysql指定的表中将数据导入到mongodb中

#源表必须有两个字段&＃xff1a;id&＃xff0c;自增&＃xff0c; mmm_ts, 时间戳

#导数据的时候先按上次操作截止的时间戳获取指定数量的数据&＃xff0c;然后再将数据存入到目标库中(如果id相同则更新数据)

#导入完毕更新截止点

class MMM

def initialize(mysql_server, mongo_server, operation_type)

&＃64;mysql_server&＃61; mysql_server

&＃64;mongo_server &＃61; mongo_server

&＃64;operation_type &＃61; operation_type

end

def migrate(source_db, source_table, process_items_each_time, target_db, target_table)

mongodb &＃61; Mongo::Connection.new(&＃64;mongo_server[:server], &＃64;mongo_server[:port]).db(target_db)

mongo_collection &＃61; mongodb.collection(target_table)

&＃64;mysql_server[:database] &＃61; source_db

ActiveRecord::Base.establish_connection(&＃64;mysql_server)

model &＃61; Class.new(ActiveRecord::Base) do

set_table_name source_table

end

operation_point &＃61; OperationPoint.new(mongodb)

opt &＃61; operation_point.get(&＃64;operation_type)

if opt &＃61;&＃61; nil

opt &＃61; 0

mongo_collection.create_index("id", :unique &＃61;> true)

else

opt &＃61; opt["point"].getlocal()

end

cnt &＃61; model.count( :conditions &＃61;> ["mmm_ts&＃61;?", opt])

#防止在过多mmm_ts字段的值都一样的情况下无法进行到下一步&＃xff0c;这里一定要避免大量mmm_ts字段的值相同

process_items_each_time &＃61; cnt &＃43; 1 if cnt >&＃61; process_items_each_time

records &＃61; model.find(:all, :conditions &＃61;> ["mmm_ts>&＃61;?", opt], :order &＃61;> "mmm_ts", :limit &＃61;> process_items_each_time)

puts("#{records.length} records read from mysql")

records.each do |record|

record_hash &＃61; {}

model.column_names.each do |column|

record_hash[column] &＃61; record.read_attribute(column)

end

mongo_collection.update({:id &＃61;> record.id}, record_hash, :upsert &＃61;> true)

opt &＃61; record.mmm_ts

end

operation_point.save(&＃64;operation_type, opt)

puts("#{records.length} records saved to mongodb")

end

end

#操作点

class OperationPoint

def initialize(db)

&＃64;table &＃61; db.collection("operation_point")

&＃64;table.create_index("operation", :unique &＃61;> true)

end

def save(operation, point)

&＃64;table.update({:operation &＃61;> operation}, {:operation &＃61;> operation, :point &＃61;> point}, :upsert &＃61;> true)

end

def get(operation)

return &＃64;table.find_one({:operation &＃61;> operation})

end

end

mmm &＃61; MMM.new(mysql_server[:main_db], mongo_server[:local], "export_mdb")

while (true) do

begin

mmm.migrate("mdb_production", "bet_plans", 3000, "mdb_production", "export_mdb")

rescue Exception &＃61;> ex

puts ex.message

puts ex.backtrace.join("\n")

sleep(30)

ensure

sleep(0.001)

end

end

分享到&＃xff1a;

2011-05-21 12:01

浏览 2089

评论