clickhouse parse acces log

load_access_log.pl

#!/usr/bin/perl

use strict;
use warnings;

use Date::Parse;
use DateTime;

my $file = $ARGV[0];
my $hostname = $ARGV[1];
print $file, "\n";

open FILE, "zcat $file|" or die $!;

open(PIPE2,"|/usr/bin/clickhouse-client --database=default --query='INSERT INTO access_log FORMAT Values'");

my $i = 0;
my @parts = ();
while (my $line = <FILE>){
 if ($line =~ /(https?):\/\/(.*?) (\d+\.\d+\.\d+\.\d+).*?\[(.*?)\] "(\w+) (.*?) HTTP.*?" (\d+) (\d+) "(.*?)" "(.*?)".*? (\d+\.\d+)$/){
      my ( $schema, $domain, $ip, $date, $method, $path, $code, $size, $ref, $ua, $resp_time)  = 
                                             ($1, $2, $3 , $4, $5, $6,$7, $8, $9,$10,$11);

      $ua =~ s/'/\\'/ig; 
      $path =~ s/'/\\'/ig; 
      $ref =~ s/'/\\'/ig; 

      my $epoch = str2time($date);
      my $datetime = DateTime->from_epoch(epoch => $epoch);

      my ($event_date, $tmp) = split('T', $datetime );
      $datetime = join(' ', $event_date, $tmp);

      push @parts, "(" .  join(',', ( "'$event_date'", "'$hostname'", "'$schema'", "'$domain'",
           "'$ip'", "'$datetime'", "'$method'", "'$path'", $code, $size, "'$ref'", "'$ua'", $resp_time)) . ")"; 
      $i++;
 }

  if ($i % 10000 == 0){
    print PIPE2 join(', ', @parts) ;
    @parts = ();
    print "insert $i rows\n";
  }
}

print PIPE2 join(', ', @parts) ;