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);
use File::Path qw(make_path remove_tree);

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


my $path = "/tmp/sack/$name";
remove_tree $path;
make_path $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();
my $total_bytes;

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 $end_t = ( $total / $rec_s ) - $dt;
        my $shard_size = -s $shard_path;

        warn sprintf "shard %s %d bytes\t%8.2f%% %8.2f/s  ETA %d:%02ds\n"
                , $shard_path, $shard_size
                , $pos * 100 / $total
                , $rec_s,
                , $end_t / 60, $end_t % 60
                ;

        $total_bytes += $shard_size;

        $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;

warn "sharded $path $total_bytes bytes\n"; 
1	#!/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	use Time::HiRes qw(time);
11	use File::Path qw(make_path remove_tree);
12
13	my $name = 'pxelator';
14	my $shard_size = 5000;
15
16
17	my $path = "/tmp/sack/$name";
18	remove_tree $path;
19	make_path $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	$total =~ s{^.*total_rows\D+(\d+).+$}{$1};
50	warn "# skip total: $total\n";
51
52	our $shard;
53	our $shard_nr = 0;
54	my $t_start = time();
55	my $total_bytes;
56
57	sub save_shard {
58	my $shard_path = sprintf("%s/%06d.%d", $path, $shard_nr++ * $shard_size, $shard_size);
59	Storable::store( $shard, $shard_path );
60
61	my $dt = time() - $t_start;
62	my $pos = $shard_nr * $shard_size;
63	my $rec_s = $pos / $dt;
64	my $end_t = ( $total / $rec_s ) - $dt;
65	my $shard_size = -s $shard_path;
66
67	warn sprintf "shard %s %d bytes\t%8.2f%% %8.2f/s ETA %d:%02ds\n"
68	, $shard_path, $shard_size
69	, $pos * 100 / $total
70	, $rec_s,
71	, $end_t / 60, $end_t % 60
72	;
73
74	$total_bytes += $shard_size;
75
76	$shard = [];
77	}
78
79	while(<$sock>) {
80	if ( /"id":"([^"]+)"/ ) {
81	# warn "get $1\n";
82	print $docs "GET /$name/$1 HTTP/1.0\r\nConnection: Keep-Alive\r\n\r\n";
83	get_chunk($docs);
84	my $json = from_json( <$docs> );
85	push @$shard, $json;
86	save_shard if $#{ $shard } == $shard_size;
87	} else {
88	warn "UNKNOWN: $_";
89	}
90	}
91	save_shard;
92
93	warn "sharded $path $total_bytes bytes\n";