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 our $join_more = 0;
  62
  63 sub join_table {
  64         $sth_join = $dbh->prepare(qq{
  65                 select
  66                         $table2.*
  67                 from $table2
  68                 where $fk > ?
  69                 order by $fk asc
  70                 limit $join_limit
  71                 offset $join_offset
  72         });
  73         print STDERR "$join_offset";
  74         $sth_join->execute( $last_id );
  75         warn "# $table2 columns ",dump( $sth_join->{NAME} );
  76         print "join ",$sth_join->rows," from $table2 offset $join_offset limit $join_limit\n";
  77         $join_more = $sth_join->rows == $join_limit ? 1 : 0;
  78 }
  79
  80 our $row_join;
  81
  82 sub fetch_row_join {
  83         $row_join = $sth_join->fetchrow_hashref();
  84         if ( ! $row_join && $join_more ) {
  85                 $join_offset += $join_limit;
  86                 join_table;
  87                 $row_join = $sth_join->fetchrow_hashref();
  88         }
  89         return $row_join;
  90 }
  91
  92 sub guess_types {
  93         my $row = shift;
  94         map { $row->{$_} * 1 } grep { defined $row->{$_} && $row->{$_} =~ /^\d+$/ } keys %$row;
  95         return $row;
  96 }
  97
  98 join_table;
  99 fetch_row_join;
 100
 101 while (my $row = $sth->fetchrow_hashref() ) {
 102
 103         while ( $row_join && $row_join->{$fk} < $row->{$pk} ) {
 104                 fetch_row_join;
 105         }
 106
 107         while ( $row_join && $row_join->{$fk} == $row->{$pk} ) {
 108                 push @{ $row->{ $table2 } }, guess_types($row_join);
 109                 print STDERR "j";
 110                 fetch_row_join;
 111         }
 112
 113         $coll->insert( guess_types($row) );
 114         print STDERR ".";
 115
 116 }
 117
 118 print STDERR "\n";