利用Moose寫Pileline模塊

aef25u

眾所周知，Spark的ML Pipelines類庫用于構(gòu)建機(jī)器學(xué)習(xí)的工作流，每一個PipelineStage 都會完成一個任務(wù)，如數(shù)據(jù)集處理轉(zhuǎn)化，模型訓(xùn)練，參數(shù)設(shè)置或數(shù)據(jù)預(yù)測等。

Spark的ML Pipelines工作流大概是這個樣子的。

1. val pipeline = new   
   
2. Pipeline().setStages(Array(labelIndexer,vectorAssembler,rfClassifier,labelConverter))  
   
3. val model = pipeline.fit(trainingData)

復(fù)制代碼

而perl的cpan上也有個Pipeline模塊，作者是用perl的舊版oop實(shí)現(xiàn)的，在分析了源碼和個人需求后，用Moose改寫出了自已的Pipeline，主要實(shí)現(xiàn)了類ML Pipelines的DataFrame（改用perl的Data::Table模塊）與Transformer功能。以下介紹如何使用Moose寫Pipeline模塊。
一、UML類圖設(shè)計(jì)


二、模塊間關(guān)系說明

• Pipeline類的dispatcher屬性是PipelineDispatch的實(shí)例對象
• PipelineDispatch類繼承自ipelineBase，屬性segments是利用Moose的屬性委托功能實(shí)現(xiàn)的用于存放PipelineSegment實(shí)例對象的數(shù)組引用
• PipelineDispatch類內(nèi)，屬性dfhash是利用Moose的屬性委托功能實(shí)現(xiàn)的用于存放各PipelineSegment->dispatch()返回值的hash引用:{ref($PipelineSegment)=>$df}即(YouSegmentClassName=>$df)
• PipelineSegment類的store屬性是Pipelinestore的實(shí)例對象
• Pipelinestore類的功能，主要是在不同的PipelineSegment間存入數(shù)據(jù)或取出數(shù)據(jù)
  

dispatch()是核心方法，抽象方法如下：

• dispatch()屬Pipeline類方法
• dispatch_loop()屬Pipeline類方法，是dispatch()的內(nèi)置方法
  
  • next()屬PipelineDispatch類方法
    

內(nèi)部調(diào)用關(guān)系如下:

1.     Pipeline->dispatch(
   
2.          Pipeline->dispatch_loop(
   
3.                Pipeline::Dispatch->next(
   
4.                       Pipeline::Segment->prepare_dispatch(Pipeline);
   
5.                       my $df = Pipeline::Segment->dispatch();
   
6.               );
   
7.           );
   
8.     );

復(fù)制代碼

注：自已寫的繼承自Pipeline::Segment的Segment類，即是Spark的ML Pipelines類的一個個Transformer

三、模塊源碼3.1Pipeline類

1. package Pipeline;
   
2. 
   
3. use Moose;
   
4. 
   
5. #use namespace::clean;
   
6. use Pipeline::Dispatch;
   
7. 
   
8. has 'debug' => (
   
9.     is      => 'rw',
   
10.     isa     => 'Int',
    
11.     default => 0,
    
12. );
    
13. has 'dispatcher' => (
    
14.     is      => 'ro',
    
15.     isa     => 'Pipeline::Dispatch',
    
16.     default => sub { Pipeline::Dispatch->new(); },
    
17.     handles => {
    
18.         get_segment => 'get',
    
19.         add_segment => 'add',
    
20.         del_segment => 'delete'
    
21.     }
    
22. );
    
23. 
    
24. has 'store' => (
    
25.     is      => 'rw',
    
26.     isa     => 'Pipeline::Store',
    
27.     default => sub { Pipeline::Store->new() },
    
28. );
    
29. 
    
30. sub segments {
    
31.     my $self = shift;
    
32.     return $self->{dispatcher}->segments(@_);
    
33. }
    
34. 
    
35. sub dispatch {
    
36.     my $self = shift;
    
37.     $self->dispatch_loop();
    
38.     $self->{dispatcher}->reset();
    
39. }
    
40. 
    
41. sub dispatch_loop {
    
42.     my $self = shift;
    
43.     $self->{dispatcher}->debug( $self->{debug} );
    
44.     while ( $self->{dispatcher}->segment_available ) {
    
45.         $self->{dispatcher}->next($self);
    
46.     }
    
47. }
    
48. sub getDf {
    
49.     my ($self,$segname) = @_;
    
50.    
    
51.     $self->{dispatcher}->getDf($segname);
    
52. }
    
53. 
    
54. #__PACKAGE__->meta->make_immutable;
    
55. 
    
56. 1;

復(fù)制代碼

3.2Pipeline:ispatch類

1. package Pipeline::Dispatch;
   
2. 
   
3. use Moose;
   
4. use Pipeline::Store;
   
5. use Pipeline::Segment;
   
6. extends 'Pipeline::Base';
   
7. 
   
8. use Data::Printer;
   
9. 
   
10. has 'segments' => (
    
11.     traits  => ['Array'],
    
12.     is      => 'rw',
    
13.     isa     => 'ArrayRef[Pipeline::Segment]',
    
14.     default => sub { [] },
    
15.     handles => {
    
16.         get               => 'get',
    
17.         add               => 'push',
    
18.         get_next_segment  => 'shift',
    
19.         delete            => 'delete',
    
20.         segment_available => 'count'
    
21.     }
    
22. );
    
23. 
    
24. has 'dispatched_segments' => (
    
25.     is      => 'rw',
    
26.     isa     => 'ArrayRef[Pipeline::Segment]',
    
27.     default => sub { [] }
    
28. );
    
29. has 'dfhash' => (
    
30.     traits  => ['Hash'],
    
31.     is      => 'ro',
    
32.     isa     => 'HashRef',
    
33.     default => sub { {} },
    
34.     handles => {
    
35.         _set_opt => 'set',
    
36.         getDf    => 'get',
    
37.     }
    
38. );
    
39. 
    
40. sub setDf {
    
41.     my ( $self, $obj, $df ) = @_;
    
42. 
    
43.     if ( defined($obj) ) {
    
44.         $self->_set_opt( ref($obj), $df );
    
45.     }
    
46.     return $self;
    
47. 
    
48. }
    
49. 
    
50. sub next {
    
51.     my $self = shift;
    
52.     my $pipe = shift;
    
53. 
    
54.     my $segment = $self->get_next_segment();
    
55.     $segment->prepare_dispatch($pipe);
    
56.     $self->emit( "dispatching to " . ref($segment) ) if $self->debug;
    
57. 
    
58.     my $df = $segment->dispatch();
    
59. 
    
60.     #將segment->dispatch的返回值克隆一份保存進(jìn)dfhash
    
61.     if ( ref($df) eq 'Data::Table' ) {
    
62.         $self->setDf( $segment, $df->clone() );
    
63.     }
    
64.     else {
    
65.         $self->setDf( $segment, $df );
    
66.         $df=undef;
    
67.     }
    
68.     push @{ $self->{dispatched_segments} }, $segment;
    
69. }
    
70. 
    
71. sub reset {
    
72.     my $self = shift;
    
73.     $self->segments( $self->{dispatched_segments} );
    
74.     $self->dispatched_segments( [] );
    
75. }
    
76. 
    
77. #__PACKAGE__->meta->make_immutable;
    
78. 
    
79. 1;

復(fù)制代碼

3.3Pipeline::Base類

1. package Pipeline::Base;
   
2. 
   
3. use Moose;
   
4. #use namespace::clean;
   
5. 
   
6. has 'debug' => (
   
7.     is      => 'rw',
   
8.     isa     => 'Int',
   
9.     default => 0,
   
10. );
    
11. 
    
12. sub emit {
    
13.     my ( $self, $mesg ) = @_;
    
14.     $self->log( $self->_format_message($mesg) ) if $self->debug;
    
15. }
    
16. 
    
17. sub log {
    
18.     my ( $self, $mesg ) = @_;
    
19.     print STDERR $mesg;
    
20. }
    
21. 
    
22. sub _format_message {
    
23.     my ( $self, $mesg ) = @_;
    
24.     my $class = ref($self);
    
25.     return "[$class] $mesg\n";
    
26. }
    
27. 
    
28. #__PACKAGE__->meta->make_immutable;
    
29. 
    
30. 1;

復(fù)制代碼

3.4Pipeline::Segment類

1. package Pipeline::Segment;
   
2. 
   
3. use Moose;
   
4. 
   
5. has 'store' => (
   
6.     is      => 'rw',
   
7.     isa     => 'Pipeline::Store',
   
8.     default => sub { Pipeline::Store->new() },
   
9. );
   
10. 
    
11. sub dispatch {
    
12.     my $self = shift;
    
13. }
    
14. 
    
15. sub prepare_dispatch {
    
16.     my ( $self, $pipe ) = @_;
    
17.     $self->store( $pipe->store );
    
18. }
    
19. 
    
20. #__PACKAGE__->meta->make_immutable;
    
21. 
    
22. 1;

復(fù)制代碼

3.5Pipeline::Store類

1. package Pipeline::Store;
   
2. 
   
3. use Moose;
   
4. 
   
5. has 'storehash' => (
   
6.     traits  => ['Hash'],
   
7.     is      => 'ro',
   
8.     isa     => 'HashRef[Object]',
   
9.     default => sub { {} },
   
10.     handles => {
    
11.         _set_opt => 'set',
    
12.         get      => 'get',
    
13.     }
    
14. );
    
15. 
    
16. sub set {
    
17.   my $self = shift;
    
18.   my $obj  = shift;
    
19.    if (defined( $obj )) {
    
20.     $self->_set_opt(ref($obj),$obj);
    
21.   }
    
22.   return $self;
    
23.   
    
24. }
    
25. 
    
26. #__PACKAGE__->meta->make_immutable;
    
27. 
    
28. 1;

復(fù)制代碼

四、Example
4.1example.pl

1. package MyDf;
   
2. use Moose;
   
3. 
   
4. extends 'Pipeline::Segment';
   
5. 
   
6. has 'df' => (
   
7.     is  => 'rw',
   
8.     isa => 'Data::Table',
   
9. );
   
10. 
    
11. package MyData;
    
12. use Moose;
    
13. 
    
14. extends 'Pipeline::Segment';
    
15. 
    
16. has 'df' => (
    
17.     is  => 'rw',
    
18.     isa => 'Data::Table',
    
19. );
    
20. 
    
21. sub dispatch {
    
22.     my $self = shift;
    
23.     $self->store->set( MyDf->new( df => $self->{df} ) );
    
24.     return $self->{df};
    
25. }
    
26. 
    
27. package MySeg1;
    
28. use Moose;
    
29. 
    
30. extends 'Pipeline::Segment';
    
31. 
    
32. sub dispatch {
    
33.     my $self = shift;
    
34.     my $df   = $self->store->get('MyDf');
    
35.     #MySeg1將MyDf增加了一行合計(jì)數(shù)
    
36.     $df->{df}->addRow( ['合計(jì)',8,undef], 3 );
    
37.     return $df->{df};
    
38. }
    
39. 
    
40. package MySeg2;
    
41. use Moose;
    
42. 
    
43. extends 'Pipeline::Segment';
    
44. 
    
45. sub dispatch {
    
46.     my $self = shift;
    
47. 
    
48.     my $df = $self->store->get('MyDf');
    
49.     #MySeg2將MyDf增加了一列總金額
    
50.     $df->{df}->addCol( [100,100,100,300],"total" ,3 );
    
51.     return $df->{df};
    
52. }
    
53. 
    
54. package main;
    
55. use lib './lib';
    
56. use Pipeline;
    
57. use Data::Table;
    
58. use Data::Printer;
    
59. 
    
60. my $headers = [ 'name', 'count', 'price' ];
    
61. 
    
62. my $rows = [ [ 'A', '1', '100' ],
    
63.              [ 'B', '2', '50' ],
    
64.              [ 'C', '5', '20' ]];
    
65. my $df = Data::Table->new( $rows, $headers, 0 );
    
66. 
    
67. #p $df->csv;
    
68. 
    
69. my $pipeline = Pipeline->new();
    
70. $pipeline->debug(1);
    
71. 
    
72. my $mydata = MyData->new( df => $df );
    
73. my $seg1 = MySeg1->new();
    
74. my $seg2 = MySeg2->new();
    
75. $pipeline->add_segment( $mydata, $seg1, $seg2 );
    
76. my $production = $pipeline->dispatch();
    
77. #p $pipeline->store->get('MyDf')->{df}->csv;
    
78. p $pipeline->getDf("MyData")->csv;
    
79. p $pipeline->getDf("MySeg1")->csv;
    
80. p $pipeline->getDf("MySeg2")->csv;
    
81. 
    
82. #Author blog:tianyv.github.io

復(fù)制代碼

104359176

能用 Moose 寫的代碼，用 Ruby， Python 也能寫吧。

不知道用 Java， C++ 能重寫嗎？

Java， C++， Golang 等靜態(tài)類型語言，對數(shù)據(jù)結(jié)構(gòu)有一些限制，例如數(shù)組中的元素都是同種類型，沒有嵌套結(jié)構(gòu)的數(shù)組。

zhouzhen1

Moose的Traits確實(shí)很好用啊。有時覺得Perl5是不是應(yīng)該直接把Moose捆綁到core里面，從工程角度講這樣更有利于推廣Perl5。然而現(xiàn)在CPAN上Moose/Moo/Mouse/Mo/Dios/Moxie。。。牛人們都在忙著玩輪子。。。