MT#18663 row bulk processing framework WIP #6
+import LNP_Define.cfg +workingdir folder for rollback logs +cleanup task +chained task execution +dry mode for import +refactoring txn demaraction for rowblock db operations +take out "lock table" impl completely, too dangerous +take out DateTime::Format::Excel dependency for now +allow '#' comments in Subscriber_Define.cfg and Lnp_Define.cfg +check for import sequence +referential integrity checks for Subscriber_Define.cfg +task summary messages / rowcounts +exclude and exclude exception number patterns for Subscriber_Define.cfg +ignore_unique options Change-Id: If1f5094e15828633e212b1a8d651c97816c388b3changes/28/6928/4
Rene Krenn
9 years ago
parent
52d66ca0de
commit
1c669a3a73
@ -0,0 +1,179 @@
|
||||
package NGCP::BulkProcessor::Projects::Migration::IPGallery::Dao::Lnp;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
use NGCP::BulkProcessor::Projects::Migration::IPGallery::ProjectConnectorPool qw(
|
||||
get_import_db
|
||||
|
||||
);
|
||||
#import_db_tableidentifier
|
||||
|
||||
use NGCP::BulkProcessor::SqlRecord qw(
|
||||
registertableinfo
|
||||
create_targettable
|
||||
checktableinfo
|
||||
copy_row
|
||||
|
||||
insert_stmt
|
||||
);
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
create_table
|
||||
gettablename
|
||||
check_table
|
||||
getinsertstatement
|
||||
|
||||
findby_lrncode_portednumber
|
||||
countby_lrncode_portednumber
|
||||
count_lrncodes
|
||||
);
|
||||
|
||||
my $tablename = 'lnp';
|
||||
my $get_db = \&get_import_db;
|
||||
#my $get_tablename = \&import_db_tableidentifier;
|
||||
|
||||
|
||||
my $expected_fieldnames = [
|
||||
'ported_number',
|
||||
'type',
|
||||
'lrn_code',
|
||||
];
|
||||
|
||||
my $primarykey_fieldnames = [ 'lrn_code', 'ported_number' ];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $fixtable_statements = [];
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = NGCP::BulkProcessor::SqlRecord->new($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,$indexes);
|
||||
|
||||
bless($self,$class);
|
||||
|
||||
copy_row($self,shift,$expected_fieldnames);
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub create_table {
|
||||
|
||||
my ($truncate) = @_;
|
||||
|
||||
my $db = &$get_db();
|
||||
|
||||
registertableinfo($db,$tablename,$expected_fieldnames,$indexes,$primarykey_fieldnames);
|
||||
return create_targettable($db,$tablename,$db,$tablename,$truncate,0,undef);
|
||||
|
||||
}
|
||||
|
||||
sub findby_lrncode_portednumber {
|
||||
|
||||
my ($lrncode,$portednumber,$load_recursive) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $rows = $db->db_get_all_arrayref(
|
||||
'SELECT * FROM ' .
|
||||
$table .
|
||||
' WHERE ' .
|
||||
$db->columnidentifier('lrn_code') . ' = ? ' .
|
||||
' AND ' . $db->columnidentifier('ported_number') . ' = ?'
|
||||
,$lrncode,$portednumber);
|
||||
|
||||
return buildrecords_fromrows($rows,$load_recursive);
|
||||
|
||||
}
|
||||
|
||||
sub countby_lrncode_portednumber {
|
||||
|
||||
my ($lrncode,$portednumber) = @_;
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
my $stmt = 'SELECT COUNT(*) FROM ' . $table;
|
||||
my @params = ();
|
||||
if (defined $lrncode) {
|
||||
$stmt .= ' WHERE ' . $db->columnidentifier('lrn_code') . ' = ?';
|
||||
push(@params,$lrncode);
|
||||
if (defined $portednumber) {
|
||||
$stmt .= ' AND ' . $db->columnidentifier('ported_number') . ' = ?';
|
||||
push(@params,$portednumber);
|
||||
}
|
||||
}
|
||||
|
||||
return $db->db_get_value($stmt,@params);
|
||||
|
||||
}
|
||||
|
||||
sub count_lrncodes {
|
||||
|
||||
check_table();
|
||||
my $db = &$get_db();
|
||||
my $table = $db->tableidentifier($tablename);
|
||||
|
||||
return $db->db_get_value('SELECT COUNT(DISTINCT ' .
|
||||
$db->columnidentifier('lrn_code') . ') FROM ' . $table);
|
||||
|
||||
}
|
||||
|
||||
sub buildrecords_fromrows {
|
||||
|
||||
my ($rows,$load_recursive) = @_;
|
||||
|
||||
my @records = ();
|
||||
my $record;
|
||||
|
||||
if (defined $rows and ref $rows eq 'ARRAY') {
|
||||
foreach my $row (@$rows) {
|
||||
$record = __PACKAGE__->new($row);
|
||||
|
||||
# transformations go here ...
|
||||
|
||||
push @records,$record;
|
||||
}
|
||||
}
|
||||
|
||||
return \@records;
|
||||
|
||||
}
|
||||
|
||||
sub getinsertstatement {
|
||||
|
||||
my ($insert_ignore) = @_;
|
||||
check_table();
|
||||
return insert_stmt($get_db,$tablename,$insert_ignore);
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,68 @@
|
||||
package NGCP::BulkProcessor::Projects::Migration::IPGallery::FileProcessors::LnpDefineFile;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
#use File::Basename;
|
||||
#use Cwd;
|
||||
#use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
use NGCP::BulkProcessor::Logging qw(
|
||||
getlogger
|
||||
);
|
||||
use NGCP::BulkProcessor::LogError qw(
|
||||
fileprocessingerror
|
||||
fileprocessingwarn
|
||||
);
|
||||
|
||||
use NGCP::BulkProcessor::FileProcessor;
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter NGCP::BulkProcessor::FileProcessor);
|
||||
our @EXPORT_OK = qw();
|
||||
|
||||
my $lineseparator = '\\n';
|
||||
my $fieldseparator = " +";
|
||||
my $encoding = 'UTF-8';
|
||||
|
||||
my $buffersize = 100 * 1024;
|
||||
my $threadqueuelength = 10;
|
||||
my $default_numofthreads = 3;
|
||||
#my $multithreading = 0;
|
||||
my $blocksize = 500;
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
|
||||
my $self = NGCP::BulkProcessor::FileProcessor->new(@_);
|
||||
|
||||
$self->{numofthreads} = shift // $default_numofthreads;
|
||||
$self->{line_separator} = $lineseparator;
|
||||
$self->{field_separator} = $fieldseparator;
|
||||
$self->{encoding} = $encoding;
|
||||
$self->{buffersize} = $buffersize;
|
||||
$self->{threadqueuelength} = $threadqueuelength;
|
||||
#$self->{multithreading} = $multithreading;
|
||||
$self->{blocksize} = $blocksize;
|
||||
|
||||
bless($self,$class);
|
||||
|
||||
#restdebug($self,__PACKAGE__ . ' file processor created',getlogger(__PACKAGE__));
|
||||
|
||||
return $self;
|
||||
|
||||
}
|
||||
|
||||
sub extractfields {
|
||||
my ($context,$line_ref) = @_;
|
||||
my $separator = $context->{instance}->{field_separator};
|
||||
$$line_ref =~ s/^ +//;
|
||||
$$line_ref =~ s/ +$//;
|
||||
return undef if length($$line_ref) == 0;
|
||||
return undef if $$line_ref =~ /^#/;
|
||||
my @fields = split(/$separator/,$$line_ref,-1);
|
||||
return \@fields;
|
||||
}
|
||||
|
||||
1;
|
||||
@ -1,10 +1,14 @@
|
||||
|
||||
features_define_filename = /home/rkrenn/test/Features_Define.cfg
|
||||
|
||||
#dry=0
|
||||
import_multithreading = 1
|
||||
|
||||
features_define_filename = /home/rkrenn/test/Features_Define.cfg
|
||||
features_define_import_numofthreads = 2
|
||||
|
||||
subscriber_define_filename = /home/rkrenn/test/Subscriber_Define.cfg
|
||||
subscriber_define_import_numofthreads = 2
|
||||
subscribernumer_exclude_pattern = ^3562770
|
||||
subscribernumer_exclude_exception_pattern = ^35627702770$
|
||||
|
||||
#dry=0
|
||||
lnp_define_filename = /home/rkrenn/test/LNP_Define.cfg
|
||||
lnp_define_import_numofthreads = 2
|
||||
|
||||
Loading…
Reference in new issue