Need help on how to use Perl DBI script to poll logs from oracle databases?

#!/bin/sh
#
export ORACLE_HOME=/ora-main/app/oracle/product/11.1.0.6/db
export ORACLE_SID=rmdev1
export LD_LIBRARY_PATH=$ORACLE_HOME/lib:$ORACLE_HOME
#
/usr/bin/dbipoll.pl -dbtype=oracle -server=servername1 -dbport=1521 -db=development -user=myuser -password=mypassword -table=dba_audit_trail -columns="OS_PROCESS,OS_USERNAME,USERNAME,USERHOST,ACTION,ACTION_NAME,TRANSACTIONID,SCN,SQL_BIND,SQL_TEXT,TO_C  HAR(EXTENDED_TIMESTAMP,'MM/DD/YYYY HH24:MI:SS'),EXTENDED_TIMESTAMP" -countfile="/app/oracle/product/monitor/countfile1" -countkey="EXTENDED_TIMESTAMP" -o="/opt/oracledb1.log"

#!/bin/sh
#
export ORACLE_HOME=/ora-main/app/oracle/product/11.1.0.6/db
export ORACLE_SID=rmdev2
export LD_LIBRARY_PATH=$ORACLE_HOME/lib:$ORACLE_HOME
#
/usr/bin/dbipoll.pl -dbtype=oracle -server=servername2 -dbport=1521 -db=development -user=myuser -password=mypassword -table=dba_audit_trail -columns="OS_PROCESS,OS_USERNAME,USERNAME,USERHOST,ACTION,ACTION_NAME,TRANSACTIONID,SCN,SQL_BIND,SQL_TEXT,TO_C  HAR(EXTENDED_TIMESTAMP,'MM/DD/YYYY HH24:MI:SS'),EXTENDED_TIMESTAMP" -countfile="/app/oracle/product/monitor/countfile2" -countkey="EXTENDED_TIMESTAMP" -o="/opt/oracledb2.log"

#!/usr/bin/perl
#
#
# Now accepts arguments for countkey, table, and columns.
# countkey will ideally be a timestamp field, such as EXTENDED_TIMESTAMP in AUD$.
#
# Need to deal with the following variations...
# 1 - Oracle local
# 2 - Oracle remote
# 3 - mysql local
# 4 - mysql remote
# 5 - Sybase local
# 6 - Sybase remote
# 
# Oracle connect string:	$dbh = DBI->connect("dbi:Oracle:host=$server;sid=$db;port=$dbport", $user, $passwd);
# Oracle connect string without env vars: $dbh = DBI->connect( "dbi:Oracle:$db", $username, $passwd )
# mysql connect string:		$dbh = DBI->connect("dbi:mysql:database=$db\;host=$server;port=$dbport","$user","$passwd");
# Sybase connect string:	$dbh = DBI->connect("dbi:sybase", $username, $password, $db);
		
use IO::Socket;
use strict;
use DBI;
use Getopt::Long;

our ($server,$db,$table,$columns,$user,$enpasswd,$passwd,$dbtype,$tcphost,$tcpport,
	$countfile,$countkey,$ofile,$dbport,$baseconnect,$dbh);

GetOptions(
		"dbtype=s"	=>\$dbtype,	# Type of DB (Oracle, MySQL, etc.)
		"server=s"	=>\$server,	# IP address of database server.
		"dbport=s"	=>\$dbport,	# Port for connecting to remote DB.
		"db=s"		=>\$db,		# Database to retriever records from.
		"user=s"	=>\$user,	# DB username.
		"password=s"	=>\$passwd,	# Unencrypted password.
		"xpassword=s"	=>\$enpasswd,	# Encrypted password.
		"table=s"	=>\$table,	# Table to retrieve from.
		"columns=s"	=>\$columns,	# Columns to fetch.
		"tcphost=s"	=>\$tcphost,	# Splunk server IP.
		"tcpport=s"	=>\$tcpport,	# Splunk tcp listener port.
		"countfile=s"	=>\$countfile,	# File to store row count status.
		"countkey=s"	=>\$countkey,	# Field to use as count value
		"o=s"		=>\$ofile,	# File for output (default is stdout).
	);

# Deal with encrypted passwords first...
if (!$passwd) {
	if ($enpasswd) {
		$passwd = `echo $enpasswd | openssl bf -d -a -pass file:key`;
	}
}

if ($dbtype =~ /oracle/i) {
	$ENV{'DBI_DRIVER'} = "Oracle";
	$dbtype = "Oracle";
	$dbh->{InactiveDestroy} = 1;
	#$baseconnect = DBI->connect("dbi:$dbtype:$db");
if (!$dbport) {
	# Assume a local DB.
	$dbh = DBI->connect("dbi:$dbtype:$db", "$user", "$passwd") ||
	  die( $DBI::errstr . "\n" );
	
} else {
	if (!$passwd) {
		$dbh = DBI->connect("dbi:Oracle:host=$server;sid=$db;port=$dbport", $user) ||
		  die( $DBI::errstr . "\n" );
	} else {
		$dbh = DBI->connect("dbi:Oracle:host=$server;sid=$db;port=$dbport", $user, $passwd) ||
		  die( $DBI::errstr . "\n" );
	}
}
	
}
if ($dbtype =~ /mysql/i) {
	$ENV{'DBI_DRIVER'} = "mysql";
	$dbtype = "mysql";
	#$baseconnect = DBI->connect("dbi:$dbtype:database=$db;host=$server");
if (!$dbport) {
	# Assume a local DB.
	$dbh = DBI->connect("dbi:$dbtype:host=$server;database=$db","$user","$passwd") ||
	  die( $DBI::errstr . "\n" );
	#$dbh = $baseconnect . "$user" . "$passwd";
} else {
	if (!$passwd) {
		$dbh = DBI->connect("dbi:$dbtype:database=$db;host=$server;port=$dbport","$user") ||
		  die( $DBI::errstr . "\n" );
	} else {
		$dbh = DBI->connect("dbi:$dbtype:database=$db;host=$server;port=$dbport","$user","$passwd") ||
		  die( $DBI::errstr . "\n" );
	}
}

}
if (($dbtype =~ /sybase/i) && (!$ENV{'SYBASE'})) {
	$ENV{'SYBASE'} = '/usr/local/freetds';
	$ENV{'DSQUERY'} = "$server";
	$dbtype = "sybase";
	#$baseconnect = DBI->connect("dbi:$dbtype", "$db");
if (!$dbport) {
	# Assume a local DB.
	$dbh = DBI->connect("dbi:$dbtype:database=$db","$user","$passwd") ||
	  die( $DBI::errstr . "\n" );
} else {
	if (!$passwd) {
		$dbh = DBI->connect("dbi:$dbtype:database=$db;port=$dbport","$user") ||
		  die( $DBI::errstr . "\n" );
	} else {
		$dbh = DBI->connect("dbi:$dbtype:database=$db;port=$dbport","$user","$passwd") ||
		  die( $DBI::errstr . "\n" );
	}
}
}
my $sth;
if ( -f $countfile) {
	# Compare vs. file with saved count
	open (CF, "<$countfile"); 
	my $filecount = readline CF;
	chomp $filecount;
	close (CF);
		# Adding an ORDER BY clause here may be a requirement
		$sth = $dbh->prepare("SELECT $columns FROM $table where $countkey > \'$filecount\' order by $countkey");
} else {
	$sth = $dbh->prepare("SELECT $columns FROM $table order by $countkey");
}

our $handle;
if (($tcphost) && ($tcpport)) {
# Send directly to splunk server...
    # Create tcp socket to send the data to the splunk server
    my $remote = IO::Socket::INET->new( Proto     => "tcp",
                                        PeerAddr  => $tcphost,
                                        PeerPort  => $tcpport,
                                        Type      => SOCK_STREAM)
                                        or die "cannot connect to tcp daemon on $tcphost";
    $remote->autoflush(1);
	$handle = $remote;
	#$target = "\$remote";
} elsif ($ofile) {
# Write to local outfile...
    open (OFILE, ">$ofile");
	$handle = *OFILE;
	#$target = *OFILE;
} else {
# send to stdout
	open (STDOUT, ">&1");
	$handle = *STDOUT;
	#$target = *STDOUT;
}

# Get some Rows...
$sth->execute or die $sth->errstr;
open (CF, ">$countfile");
while(my $hash_ref = $sth->fetchrow_hashref) {
  my $output = "";
  my $lastrow;
  foreach my $key (keys(%$hash_ref)) {
      my $str = $hash_ref->{$key};
      if ($str =~ /"/) {
	  (my $xstr = $str) =~ s/"/\\"/g;
	  $str = "\"$xstr\"";
      } else {
	  $str = "\"$str\"" if ($str =~ / /);
	  $str = "\"$str\"" if ($str =~ /,/);
      }
      $output = $output . $key . "=" . $str . ",";
      $lastrow = $hash_ref->{$key} if (lc($key) eq lc($countkey));
      
  }
  chop($output);
   # Update countfile
      # Where did we leave off?...
      *CF->autoflush(1);
      seek(CF,0,0);
      print CF "$lastrow\n";

# And send them somewhere.
print $handle $output . "\n";
}

$sth->finish();
$dbh->disconnect;

close (CF);
close $handle;