MT#18663 row bulk processing framework WIP #4
+"import" stage continued: +Dao's/db schema for the sqlite db to map the Features_Define.cfg content +project-specific connector pool +Importing Features_Define.cfg is working, 1.2+4.2 mio. records can be inserted into the temporary sqlite db in 4min40sec using 4 hyperthread cpus +importing melita's remaining files from the set will be less complicated by far +GetOpt setup +a stage is a task, multiple tasks can be passed as args, to be executed at once +dry mode option +force option +project settings file extended Change-Id: I6a84eb90756ad382dc3e96636220254b5e3b5023changes/06/6906/1
Rene Krenn
10 years ago
parent
0d40b0c4ff
commit
4570a01bbc
@ -0,0 +1,125 @@
|
||||
package Projects::Migration::IPGallery::Dao::FeatureOption;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use File::Basename;
|
||||
use Cwd;
|
||||
use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
#use Logging qw(getlogger);
|
||||
|
||||
use Projects::Migration::IPGallery::ProjectConnectorPool qw(
|
||||
get_import_db
|
||||
|
||||
);
|
||||
#import_db_tableidentifier
|
||||
|
||||
use SqlRecord qw(
|
||||
registertableinfo
|
||||
create_targettable
|
||||
checktableinfo
|
||||
copy_row
|
||||
|
||||
insert_stmt
|
||||
);
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
create_table
|
||||
gettablename
|
||||
check_table
|
||||
getinsertstatement
|
||||
|
||||
test_table_bycolumn1
|
||||
test_table_local_select
|
||||
test_table_source_select
|
||||
test_table_source_select_temptable
|
||||
);
|
||||
|
||||
my $tablename = 'feature_option';
|
||||
my $get_db = \&get_import_db;
|
||||
#my $get_tablename = \&import_db_tableidentifier;
|
||||
|
||||
|
||||
my $expected_fieldnames = [ 'subscribernumber',
|
||||
'option'];
|
||||
|
||||
my $primarykey_fieldnames = [ 'subscribernumber', 'option' ];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $fixtable_statements = [];
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = 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 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 {
|
||||
|
||||
check_table();
|
||||
return insert_stmt($get_db,$tablename);
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,126 @@
|
||||
package Projects::Migration::IPGallery::Dao::FeatureOptionSet;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use File::Basename;
|
||||
use Cwd;
|
||||
use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../../');
|
||||
|
||||
#use Logging qw(getlogger);
|
||||
|
||||
use Projects::Migration::IPGallery::ProjectConnectorPool qw(
|
||||
get_import_db
|
||||
|
||||
);
|
||||
#import_db_tableidentifier
|
||||
|
||||
use SqlRecord qw(
|
||||
registertableinfo
|
||||
create_targettable
|
||||
checktableinfo
|
||||
copy_row
|
||||
|
||||
insert_stmt
|
||||
);
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter SqlRecord);
|
||||
our @EXPORT_OK = qw(
|
||||
create_table
|
||||
gettablename
|
||||
check_table
|
||||
getinsertstatement
|
||||
|
||||
test_table_bycolumn1
|
||||
test_table_local_select
|
||||
test_table_source_select
|
||||
test_table_source_select_temptable
|
||||
);
|
||||
|
||||
my $tablename = 'feature_option_set';
|
||||
my $get_db = \&get_import_db;
|
||||
#my $get_tablename = \&import_db_tableidentifier;
|
||||
|
||||
|
||||
my $expected_fieldnames = [ 'subscribernumber',
|
||||
'option',
|
||||
'optionsetitem' ];
|
||||
|
||||
my $primarykey_fieldnames = [ 'subscribernumber', 'option', 'optionsetitem' ];
|
||||
|
||||
my $indexes = {};
|
||||
|
||||
my $fixtable_statements = [];
|
||||
|
||||
sub new {
|
||||
|
||||
my $class = shift;
|
||||
my $self = 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 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 {
|
||||
|
||||
check_table();
|
||||
return insert_stmt($get_db,$tablename);
|
||||
|
||||
}
|
||||
|
||||
sub gettablename {
|
||||
|
||||
return $tablename;
|
||||
|
||||
}
|
||||
|
||||
sub check_table {
|
||||
|
||||
return checktableinfo($get_db,
|
||||
$tablename,
|
||||
$expected_fieldnames,
|
||||
$indexes);
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,85 @@
|
||||
package Projects::Migration::IPGallery::ProjectConnectorPool;
|
||||
use strict;
|
||||
|
||||
## no critic
|
||||
|
||||
use File::Basename;
|
||||
use Cwd;
|
||||
use lib Cwd::abs_path(File::Basename::dirname(__FILE__) . '/../../../');
|
||||
|
||||
use Projects::Migration::IPGallery::Settings qw(
|
||||
$import_db_file
|
||||
);
|
||||
|
||||
use ConnectorPool qw(
|
||||
get_connectorinstancename
|
||||
);
|
||||
|
||||
#use SqlConnectors::MySQLDB;
|
||||
#use SqlConnectors::OracleDB;
|
||||
#use SqlConnectors::PostgreSQLDB;
|
||||
use SqlConnectors::SQLiteDB qw(
|
||||
$staticdbfilemode
|
||||
cleanupdbfiles
|
||||
);
|
||||
#use SqlConnectors::CSVDB;
|
||||
#use SqlConnectors::SQLServerDB;
|
||||
#use RestConnectors::NGCPRestApi;
|
||||
|
||||
use SqlRecord qw(cleartableinfo);
|
||||
|
||||
#use Utils qw(threadid);
|
||||
|
||||
require Exporter;
|
||||
our @ISA = qw(Exporter);
|
||||
our @EXPORT_OK = qw(
|
||||
get_import_db
|
||||
import_db_tableidentifier
|
||||
|
||||
destroy_dbs
|
||||
);
|
||||
|
||||
# thread connector pools:
|
||||
my $import_dbs = {};
|
||||
|
||||
|
||||
sub get_import_db {
|
||||
|
||||
my ($instance_name,$reconnect) = @_;
|
||||
my $name = get_connectorinstancename($instance_name); #threadid(); #shift;
|
||||
|
||||
if (not defined $import_dbs->{$name}) {
|
||||
$import_dbs->{$name} = SqlConnectors::SQLiteDB->new($instance_name); #$name);
|
||||
if (not defined $reconnect) {
|
||||
$reconnect = 1;
|
||||
}
|
||||
}
|
||||
if ($reconnect) {
|
||||
$import_dbs->{$name}->db_connect($staticdbfilemode,$import_db_file);
|
||||
}
|
||||
|
||||
return $import_dbs->{$name};
|
||||
|
||||
}
|
||||
|
||||
sub import_db_tableidentifier {
|
||||
|
||||
my ($get_target_db,$tablename) = @_;
|
||||
my $target_db = (ref $get_target_db eq 'CODE') ? &$get_target_db() : $get_target_db;
|
||||
return $target_db->getsafetablename(SqlConnectors::SQLiteDB::get_tableidentifier($tablename,$staticdbfilemode,$import_db_file));
|
||||
|
||||
}
|
||||
|
||||
|
||||
sub destroy_dbs {
|
||||
|
||||
|
||||
foreach my $name (keys %$import_dbs) {
|
||||
cleartableinfo($import_dbs->{$name});
|
||||
undef $import_dbs->{$name};
|
||||
delete $import_dbs->{$name};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
1;
|
||||
@ -0,0 +1,39 @@
|
||||
##general settings:
|
||||
working_path = /var/sipwise/Migration/IPGallery
|
||||
cpucount = 4
|
||||
enablemultithreading = 1
|
||||
|
||||
##gearman/service listener config:
|
||||
jobservers = 127.0.0.1:4730
|
||||
|
||||
##NGCP MySQL connectivity - "accounting" db:
|
||||
accounting_host = 127.0.0.1
|
||||
accounting_port = 3306
|
||||
accounting_databasename = accounting
|
||||
accounting_username = root
|
||||
accounting_password =
|
||||
|
||||
##NGCP MySQL connectivity - "billing" db:
|
||||
billing_host = 127.0.0.1
|
||||
billing_port = 3306
|
||||
billing_databasename = accounting
|
||||
billing_username = root
|
||||
billing_password =
|
||||
|
||||
##NGCP REST-API connectivity:
|
||||
ngcprestapi_uri = https://127.0.0.1:443
|
||||
ngcprestapi_username = administrator
|
||||
ngcprestapi_password = administrator
|
||||
ngcprestapi_realm = api_admin_http
|
||||
|
||||
##sending email:
|
||||
emailenable = 0
|
||||
erroremailrecipient =
|
||||
warnemailrecipient =
|
||||
completionemailrecipient = rkrenn@sipwise.com
|
||||
successemailrecipient =
|
||||
|
||||
##logging:
|
||||
fileloglevel = OFF
|
||||
screenloglevel = INFO
|
||||
emailloglevel = OFF
|
||||
@ -1 +0,0 @@
|
||||
working_path = /var/sipwise/Migration/IPGallery
|
||||
@ -1 +1,7 @@
|
||||
x = y
|
||||
|
||||
features_define_filename = /home/rkrenn/test/Features_Define.cfg
|
||||
|
||||
import_multithreading = 1
|
||||
feature_define_import_numofthreads = 10
|
||||
|
||||
#dry=0
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -1 +1,39 @@
|
||||
#working_path = /var/sipwise
|
||||
##general settings:
|
||||
#working_path = /var/sipwise
|
||||
cpucount = 1
|
||||
enablemultithreading = 1
|
||||
|
||||
##gearman/service listener config:
|
||||
jobservers = 127.0.0.1:4730
|
||||
|
||||
##NGCP MySQL connectivity - "accounting" db:
|
||||
accounting_host = 127.0.0.1
|
||||
accounting_port = 3306
|
||||
accounting_databasename = accounting
|
||||
accounting_username = root
|
||||
accounting_password =
|
||||
|
||||
##NGCP MySQL connectivity - "billing" db:
|
||||
billing_host = 127.0.0.1
|
||||
billing_port = 3306
|
||||
billing_databasename = accounting
|
||||
billing_username = root
|
||||
billing_password =
|
||||
|
||||
##NGCP REST-API connectivity:
|
||||
ngcprestapi_uri = https://127.0.0.1:443
|
||||
ngcprestapi_username = administrator
|
||||
ngcprestapi_password = administrator
|
||||
ngcprestapi_realm = api_admin_http
|
||||
|
||||
##sending email:
|
||||
emailenable = 0
|
||||
erroremailrecipient =
|
||||
warnemailrecipient =
|
||||
completionemailrecipient = rkrenn@sipwise.com
|
||||
successemailrecipient =
|
||||
|
||||
##logging:
|
||||
fileloglevel = OFF
|
||||
screenloglevel = INFO
|
||||
emailloglevel = OFF
|
||||
|
||||
Loading…
Reference in new issue