Perl访问MSSQL并迁移到MySQL数据库脚本实例

die"Usage:qqdb\n"if@ARGV!=1;
my$db=$ARGV[0];

print"Connectintodatabases$db...\n";
my$cs="DRIVER={FreeTDS};SERVER=MSSQL的服务器;PORT=1433;DATABASE=$db;UID=sa;PWD=MSSQL密码;TDS_VERSION=7.1;charset=gb2312";

subdb_connect
{
   my$src=DBI->connect("dbi:ODBC:$cs")ordie$@;
   my$target=DBI->connect("dbi:mysql:host=MySQL服务器","MySQL用户名","MySQL密码")ordie$@;
   return($src,$target);
}
my($src,$target)=db_connect;

print"Readingtableschemas....\n";

my$q_tables=$src->prepare("SELECTnameFROMsysobjectsWHERExtype="U"ANDname!="dtproperties";");#获取所有表名
my$q_key_usage=$src->prepare("SELECTTABLE_NAME,COLUMN_NAMEfromINFORMATION_SCHEMA.KEY_COLUMN_USAGE;");#获取表的主键
$q_tables->execute;
my@tables=();
my%keys=();
push@tables,@_while@_=$q_tables->fetchrow_array;

$q_tables->finish;

$q_key_usage->execute();
$keys{$_[0]}=$_[1]while@_=$q_key_usage->fetchrow_array;
$q_key_usage->finish;

#获取表的索引信息
my$q_index=$src->prepare(qq(
   SELECTT.name,C.name
   FROMsys.index_columnsI
   INNERJOINsys.tablesTONT.object_id=I.object_id
   INNERJOINsys.columnsCONC.column_id=I.column_idANDI.object_id=C.object_id;
));
$q_index->execute;
my%table_indices=();
while(my@row=$q_index->fetchrow_array)
{
   my($table,$column)=@row;
   my$columns=$table_indices{$table};
   $columns=$table_indices{$table}=[]ifnot$columns;
   push@$columns,$column;
}
$q_index->finish;

#在目标MySQL上创建对应的数据库
$target->do("DROPDATABASEIFEXISTS`$db`;")ordie"Cannotdropolddatabase$db\n";
$target->do("CREATEDATABASE`$db`DEFAULTCHARSET=utf8COLLATEutf8_general_ci;")ordie"Cannotcreatedatabase$db\n";
$target->disconnect;
$src->disconnect;

my$total_start=time;
formy$table(@tables)
{
   my$pid=fork;
   unless($pid)
   {
       ($src,$target)=db_connect;
       my$start=time;
       $src->do("USE$db;");
       #获取表结构，用来生成MySQL用的DDL
       my$q_schema=$src->prepare("SELECTCOLUMN_NAME,IS_NULLABLE,DATA_TYPE,CHARACTER_MAXIMUM_LENGTHfromINFORMATION_SCHEMA.COLUMNSwhereTABLE_NAME=?ORDERBYORDINAL_POSITION;");
       $target->do("USE`$db`;");
       $target->do("SETNAMESutf8;");
       my$key_column=$keys{$table};
       my$ddl="CREATETABLE`$table`(\n";
       $q_schema->execute($table);
       my@fields=();
       while(my@row=$q_schema->fetchrow_array)
       {
           my($column,$nullable,$datatype,$length)=@row;
           my$field="`$column`$datatype";
           $field.="($length)"if$length;
           $field.="PRIMARYKEY"if$key_columneq$column;
           push@fields,$field;
       }
       $ddl.=join(",\n",@fields);
       $ddl.="\n)ENGINE=MyISAM;\n\n";
       $target->do($ddl)ordie"Cannotcreatetable$table\n";
       #创建索引
       my$indices=$table_indices{$table};
       if($indices)
       {
           for(@$indices)
           {
               $target->do("CREATEINDEX`$_`ON`$table`(`$_`);\n")ordie"Cannotcreateindexon$db.$table$.$_\n";
           }
       }
       #转移数据
       my@placeholders=map{"?"}@fields;
       my$insert_sql="INSERTDELAYEDINTO$tableVALUES(".(join",",@placeholders).");\n";
       my$insert=$target->prepare($insert_sql);
       my$select=$src->prepare("SELECT*FROM$table;");
       $select->execute;
       $select->{"LongReadLen"}=1000;
       $select->{"LongTruncOk"}=1;
       $target->do("SETAUTOCOMMIT=0;");
       $target->do("STARTTRANSACTION;");
       my$rows=0;
       while(my@row=$select->fetchrow_array)
       {
           $insert->execute(@row);
           $rows++;
       }
       $target->do("COMMIT;");
       #结束，输出任务信息
       my$elapsed=time-$start;
       print"Childprocess$$fortable$db.$tabledone,$rowsrecords,$elapsedseconds.\n";
       exit(0);
   }
}
print"Waitingforchildprocesses\n";
#等待所有子进程结束
while(wait()!=-1){}
my$total_elapsed=time-$total_start;
print"Alltasksfrom$dbfinished,$total_elapsedseconds.\n";