Changeset 19262

Timestamp:

09/13/08 16:15:19 (5 years ago)

Author:

lopnor

Message:

lang/perl/App-Hachero: hadoop streaming support, but not tested yet.

Location:

lang/perl/App-Hachero/trunk

Files:

• config.yaml (modified) (2 diffs)
• lib/App/Hachero.pm (modified) (4 diffs)
• lib/App/Hachero/Plugin/Analyze/AccessCount.pm (modified) (1 diff)
• lib/App/Hachero/Plugin/Input/File.pm (modified) (3 diffs)
• lib/App/Hachero/Plugin/Output/HadoopReduce.pm (added)
• lib/App/Hachero/Plugin/OutputLine (added)
• lib/App/Hachero/Plugin/OutputLine/HadoopMap.pm (added)
• lib/App/Hachero/Plugin/Parse/HadoopReduce.pm (added)
• t/plugin/analyze/02_accesscount.t (modified) (4 diffs)
• t/plugin/input/01_stdin.t (modified) (1 diff)
• t/plugin/input/03_file.t (modified) (1 diff)
• t/plugin/output/02_hadoop_reduce.t (added)
• t/plugin/output_line (added)
• t/plugin/output_line/01_hadoop_map.t (added)
• t/plugin/parse/03_hadoop_reduce.t (added)

lang/perl/App-Hachero/trunk/config.yaml

@@ -1 @@
-global:
-  work_path: t/work
-
 plugins:
 
@@ -14 +11 @@
         time_zone: Asia/Tokyo
 
-  - module: Analyze::AccessCount
-
-  - module: Output::DBIC
-    config:
-        connect_info:
-            - dbi:mysql:test
+  - module: OutputLine::HadoopMap
+#  
+#  - module: Analyze::AccessCount
+#
+#  - module: Output::DBIC
+#    config:
+#        connect_info:
+#            - dbi:mysql:test

lang/perl/App-Hachero/trunk/lib/App/Hachero.pm

@@ -18 +18 @@
 
 sub new {
-    my $class = shift;
-    my $args = $_[0];
+    my ($class,$args) = @_;
+    $class->_setup_plugins_static($args);
+    my $self = $class->SUPER::new($args);
+    $self->result({});
+    $context = $self;
+    $self;
+}
+
+sub _setup_plugins_static {
+    my ($class, $args) = @_;
     my $config = $class->setup_config( $args->{config} );
 
@@ -36 +44 @@
         $packages_from_plugin_path = $collect->modules;
     }
-
     $class->load_plugins(@plugins);
-
-    my $self = $class->SUPER::new(@_);
-    $self->result({});
-    $context = $self;
-    $self;
 }
 
@@ -53 +55 @@
 sub run {
     my $self = shift;
+    $self->log(debug => sprintf ('run start: %s', scalar localtime));
     $self->initialize;
     $self->run_hook('fetch');
@@ -59 +62 @@
         $self->run_hook_and_check('classify') or next;
         $self->run_hook_and_check('analyze') or next;
+        $self->run_hook('output_line');
     }
     $self->run_hook('associate');
     $self->run_hook('output');
+    $self->log(debug => sprintf ('run end: %s', scalar localtime));
 }
 

lang/perl/App-Hachero/trunk/lib/App/Hachero/Plugin/Analyze/AccessCount.pm

@@ -4 +4 @@
 use base qw(App::Hachero::Plugin::Base);
 use DateTime::Format::MySQL;
+use Digest::MD5;
 
 sub analyze : Hook {
     my ($self, $context, $args) = @_;
     my $req = $context->currentinfo->{request} or return;
+    my $truncate = $self->config->{config}->{truncate_to} || 'minute';
     my $time = DateTime::Format::MySQL->format_datetime(
-        $req->{datetime}->clone->truncate(to => 'minute')
+        $req->{datetime}->clone->truncate(to => $truncate)
     );
-    $context->result->{'AccessCount'}->{$time} = {
+    my $hash = Digest::MD5::md5_base64($time);
+    $context->result->{'AccessCount'}->{$hash} = {
         datetime => $time,
-        count => ($context->result->{'AccessCount'}->{$time}->{count} || 0) + 1,
+        count => ($context->result->{'AccessCount'}->{$hash}->{count} || 0) + 1,
     }
 }

lang/perl/App-Hachero/trunk/lib/App/Hachero/Plugin/Input/File.pm

@@ -4 +4 @@
 use base qw(App::Hachero::Plugin::Base);
 use File::Find::Rule;
+use File::stat;
 
 sub init {
@@ -12 +13 @@
     my ($self, $context) = @_;
     $self->{rule} ||= $self->_get_rule($context);
-    my $file;
-    while (1) {
-        $file = $self->{rule}->match or return;
-        $file !~ /^\.{1,2}/ and last;
+    my $file = $self->{rule}->match or return;
+
+    if ($context->conf->{global}->{log}->{level} eq 'debug') {
+        $context->log(debug => "opening file: $file");
+        $self->{readsize} = 0;
+        $self->{filesize} = stat($file)->size;
     }
     open my $fh, '<', $file or die;
@@ -34 +37 @@
         my $fh = $self->{fh} || $self->_fetch($context) or return;
         $line = <$fh>;
+        if ($context->conf->{global}->{log}->{level} eq 'debug') {
+            $self->{readsize}+= do {use bytes; length $line} if $line;
+            print STDERR "\rreading file: $self->{readsize}/$self->{filesize}";
+        }
         unless ($line) {
             close $self->{fh};

lang/perl/App-Hachero/trunk/t/plugin/analyze/02_accesscount.t

@@ -14 +14 @@
         {
             module => 'Analyze::AccessCount',
-            config => {
-
-            },
+            config => {truncate_to => 'hour'},
         }
     ]
@@ -35 +33 @@
     $app->run_hook('parse');
     $app->run_hook('analyze');
-    is_deeply $app->result->{'AccessCount'}, $block->expected;
+    my $value = (values %{$app->result->{AccessCount}})[0];
+    is_deeply $value, $block->expected;
 }
 
@@ -56 +55 @@
 
 --- expected
-'2008-08-07 00:00:00': 
-    datetime: '2008-08-07 00:00:00'
-    count: 1
+datetime: '2008-08-07 00:00:00'
+count: 1
 
 === test2
@@ -75 +73 @@
 
 --- expected
-'2008-08-06 15:05:00':
-    datetime: '2008-08-06 15:05:00'
-    count: 1
+datetime: '2008-08-06 15:00:00'
+count: 1
+

lang/perl/App-Hachero/trunk/t/plugin/input/01_stdin.t

@@ -9 +9 @@
 
 {
-    my $line = 'hogehoge';
+    my $line = 'hoge';
     my ($fh, $log) = File::Temp::tempfile;
     print $fh $line;

lang/perl/App-Hachero/trunk/t/plugin/input/03_file.t

@@ -10 +10 @@
 
 my $config = {
-    global => {work_path => 't/work'},
+    global => {
+        work_path => 't/work',
+        log => {level => 'error'},
+    },
     plugins => [
         {