利用Moose写Pileline模块

aef25u

众所周知，Spark的ML Pipelines类库用于构建机器学习的工作流，每一个PipelineStage 都会完成一个任务，如数据集处理转化，模型训练，参数设置或数据预测等。

Spark的ML Pipelines工作流大概是这个样子的。

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

复制代码

而perl的cpan上也有个Pipeline模块，作者是用perl的旧版oop实现的，在分析了源码和个人需求后，用Moose改写出了自已的Pipeline，主要实现了类ML Pipelines的DataFrame（改用perl的Data::Table模块）与Transformer功能。以下介绍如何使用Moose写Pipeline模块。
一、UML类图设计


二、模块间关系说明

• Pipeline类的dispatcher属性是PipelineDispatch的实例对象
• PipelineDispatch类继承自ipelineBase，属性segments是利用Moose的属性委托功能实现的用于存放PipelineSegment实例对象的数组引用
• PipelineDispatch类内，属性dfhash是利用Moose的属性委托功能实现的用于存放各PipelineSegment->dispatch()返回值的hash引用:{ref($PipelineSegment)=>$df}即(YouSegmentClassName=>$df)
• PipelineSegment类的store属性是Pipelinestore的实例对象
• Pipelinestore类的功能，主要是在不同的PipelineSegment间存入数据或取出数据
  

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

• dispatch()属Pipeline类方法
• dispatch_loop()属Pipeline类方法，是dispatch()的内置方法
  
  • next()属PipelineDispatch类方法
    

内部调用关系如下:

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.     );

复制代码

注：自已写的继承自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;

复制代码

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的返回值克隆一份保存进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;

复制代码

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;

复制代码

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;

复制代码

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;

复制代码

四、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增加了一行合计数
    
36.     $df->{df}->addRow( ['合计',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

复制代码

104359176

能用 Moose 写的代码，用 Ruby， Python 也能写吧。

不知道用 Java， C++ 能重写吗？

Java， C++， Golang 等静态类型语言，对数据结构有一些限制，例如数组中的元素都是同种类型，没有嵌套结构的数组。

zhouzhen1

Moose的Traits确实很好用啊。有时觉得Perl5是不是应该直接把Moose捆绑到core里面，从工程角度讲这样更有利于推广Perl5。然而现在CPAN上Moose/Moo/Mouse/Mo/Dios/Moxie。。。牛人们都在忙着玩轮子。。。