trunk/bin/couchdb2shards.pl

#!/usr/bin/perl

use warnings;
use strict;

use IO::Socket::INET;
use Storable qw();
use JSON;
use Data::Dump qw(dump);
use Time::HiRes qw(time);

my $name = 'pxelator';
my $shard_size = 5000;


my $path = '/tmp/sack';
mkdir $path;
$path .= '/' . $name;
mkdir $path;


sub couchdb_socket {
        IO::Socket::INET->new(
                PeerAddr => '10.60.0.91',
                PeerPort => 5984,
                Proto => 'tcp',
        ) || die $!;
}

sub get_chunk {
        my $sock = shift;
        my $chunk;
        while(<$sock>) {
                $chunk .= $_;
                last if /^[\n\r]+$/;
        }
#       warn "# $sock\n$chunk\n";
        return $chunk;
}

my $sock = couchdb_socket;
my $docs = couchdb_socket;

print $sock "GET /$name/_all_docs HTTP/1.0\r\n\r\n";

get_chunk($sock);

my $total = <$sock>;
$total =~ s{^.*total_rows\D+(\d+).+$}{$1};
warn "# skip total: $total\n";

our $shard;
our $shard_nr = 0;
my $t_start = time();

sub save_shard {
        my $shard_path = sprintf("%s/%06d.%d", $path, $shard_nr++ * $shard_size, $shard_size);
        Storable::store( $shard, $shard_path );

        my $dt = time() - $t_start;
        my $pos = $shard_nr * $shard_size;
        my $rec_s = $pos / $dt;
        my $total_t = $total / $rec_s;

        warn sprintf "shard %s %d bytes\t%3.2f%% %.2f/s ETA %d:%02ds\n"
                , $shard_path, -s $shard_path
                , $pos * 100 / $total
                , $rec_s,
                , $total_t / 60, $total_t % 60
                ;

        $shard = [];
}

while(<$sock>) {
        if ( /"id":"([^"]+)"/ ) {
#               warn "get $1\n";
                print $docs "GET /$name/$1 HTTP/1.0\r\nConnection: Keep-Alive\r\n\r\n";
                get_chunk($docs);
                my $json = from_json( <$docs> );
                push @$shard, $json;
                save_shard if $#{ $shard } == $shard_size;
        } else {
                warn "UNKNOWN: $_";
        }
}
save_shard;
1	dpavlin	201	#!/usr/bin/perl
2
3			use warnings;
4			use strict;
5
6			use IO::Socket::INET;
7			use Storable qw();
8			use JSON;
9			use Data::Dump qw(dump);
10	dpavlin	206	use Time::HiRes qw(time);
11	dpavlin	201
12			my $name = 'pxelator';
13	dpavlin	206	my $shard_size = 5000;
14	dpavlin	201
15
16			my $path = '/tmp/sack';
17			mkdir $path;
18			$path .= '/' . $name;
19			mkdir $path;
20
21
22			sub couchdb_socket {
23			IO::Socket::INET->new(
24			PeerAddr => '10.60.0.91',
25			PeerPort => 5984,
26			Proto => 'tcp',
27			) \|\| die $!;
28			}
29
30			sub get_chunk {
31			my $sock = shift;
32			my $chunk;
33			while(<$sock>) {
34			$chunk .= $_;
35			last if /^[\n\r]+$/;
36			}
37			# warn "# $sock\n$chunk\n";
38			return $chunk;
39			}
40
41			my $sock = couchdb_socket;
42			my $docs = couchdb_socket;
43
44			print $sock "GET /$name/_all_docs HTTP/1.0\r\n\r\n";
45
46			get_chunk($sock);
47
48			my $total = <$sock>;
49	dpavlin	206	$total =~ s{^.*total_rows\D+(\d+).+$}{$1};
50	dpavlin	201	warn "# skip total: $total\n";
51
52			our $shard;
53			our $shard_nr = 0;
54	dpavlin	206	my $t_start = time();
55	dpavlin	201
56			sub save_shard {
57			my $shard_path = sprintf("%s/%06d.%d", $path, $shard_nr++ * $shard_size, $shard_size);
58			Storable::store( $shard, $shard_path );
59	dpavlin	206
60			my $dt = time() - $t_start;
61			my $pos = $shard_nr * $shard_size;
62			my $rec_s = $pos / $dt;
63			my $total_t = $total / $rec_s;
64
65			warn sprintf "shard %s %d bytes\t%3.2f%% %.2f/s ETA %d:%02ds\n"
66			, $shard_path, -s $shard_path
67			, $pos * 100 / $total
68			, $rec_s,
69			, $total_t / 60, $total_t % 60
70			;
71
72	dpavlin	201	$shard = [];
73			}
74
75			while(<$sock>) {
76			if ( /"id":"([^"]+)"/ ) {
77			# warn "get $1\n";
78			print $docs "GET /$name/$1 HTTP/1.0\r\nConnection: Keep-Alive\r\n\r\n";
79			get_chunk($docs);
80			my $json = from_json( <$docs> );
81			push @$shard, $json;
82			save_shard if $#{ $shard } == $shard_size;
83			} else {
84			warn "UNKNOWN: $_";
85			}
86			}
87			save_shard;