dbi2mongo.pl

   1 #!/usr/bin/perl -w
   2
   3 sub BEGIN {
   4 $ENV{DBI_AUTOPROXY}='dbi:Gofer:transport=stream;url=ssh:dpavlin@koha.ffzg.hr';
   5 }
   6
   7 use strict;
   8 use DBI;
   9 use MongoDB;
  10 use Data::Dump qw/dump/;
  11
  12 $|++;
  13
  14 my $debug = @ARGV ? 1 : 0;
  15
  16 our ( $dbi, $user, $password ) = ( "DBI:mysql:database=test" );
  17 our ( $database, $collection ) = ( 'test', 'test' );
  18 our ( $table,  $pk ) = ( 'biblio'      => 'biblionumber' );
  19 our ( $table2, $fk ) = ( 'biblioitems' => 'biblionumber' );
  20
  21 my $limit      = 50000;
  22 my $join_limit = 10000;
  23
  24 require 'config.pl';
  25
  26 warn "# $dbi $user -> $database $collection $table.$pk<->$table2.$fk\n";
  27
  28 my $conn = MongoDB::Connection->new;
  29 my $db   = $conn->get_database( $database );
  30 my $coll = $db->get_collection( $collection );
  31 my $dbh  = DBI->connect($dbi,$user,$password, {
  32         RaiseError => 1,
  33 #       mysql_enable_utf8 => 1,
  34 });
  35
  36 $db->drop if $debug;
  37
  38 # db.items.find().sort({_id:-1}).limit(1);
  39 my $last = $coll->query()->sort({ '_id' => -1 })->limit(1)->next;
  40 warn dump( $last );
  41 my $last_id = $last->{_id} || 0;
  42
  43 print "import $table.$pk > $last_id from $dbi\n";
  44
  45 my $sth = $dbh->prepare(qq{
  46         select
  47                 $pk as _id,
  48                 $table.*
  49         from $table
  50         where $pk > ?
  51         order by $pk asc
  52         limit $limit
  53 });
  54
  55 $sth->execute( $last_id );
  56 warn "# $table columns ",dump( $sth->{NAME} );
  57 print "import ",$sth->rows," from $table\n";
  58
  59 our $join_offset = 0;
  60 our $sth_join;
  61
  62 sub join_table {
  63         $sth_join = $dbh->prepare(qq{
  64                 select
  65                         $table2.*
  66                 from $table2
  67                 where $fk > ?
  68                 order by $fk asc
  69                 limit $join_limit
  70                 offset $join_offset
  71         });
  72         print STDERR "$join_offset";
  73         $sth_join->execute( $last_id );
  74         warn "# $table2 columns ",dump( $sth_join->{NAME} );
  75         print "join ",$sth_join->rows," from $table2 offset $join_offset limit $join_limit\n";
  76 }
  77
  78 our $row_join;
  79
  80 sub fetch_row_join {
  81         $row_join = $sth_join->fetchrow_hashref();
  82         if ( ! $row_join && $sth_join->rows == $join_limit ) {
  83                 $join_offset += $join_limit;
  84                 join_table;
  85                 $row_join = $sth_join->fetchrow_hashref();
  86         }
  87         return $row_join;
  88 }
  89
  90 sub guess_types {
  91         my $row = shift;
  92         map { $row->{$_} * 1 } grep { defined $row->{$_} && $row->{$_} =~ /^\d+$/ } keys %$row;
  93         return $row;
  94 }
  95
  96 join_table;
  97 fetch_row_join;
  98
  99 while (my $row = $sth->fetchrow_hashref() ) {
 100
 101         while ( $row_join && $row_join->{$fk} < $row->{$pk} ) {
 102                 fetch_row_join;
 103         }
 104
 105         while ( $row_join && $row_join->{$fk} == $row->{$pk} ) {
 106                 push @{ $row->{ $table2 } }, guess_types($row_join);
 107                 print STDERR "j";
 108                 fetch_row_join;
 109         }
 110
 111         $coll->insert( guess_types($row) );
 112         print STDERR ".";
 113
 114 }
 115
 116 print STDERR "\n";