converters unification

Author: sunsingerus

run.sh

@@ -5,9 +5,9 @@ python3 main.py \
 	--src-host=127.0.0.1 --src-user=reader --src-password=qwerty \
 	--dst-host=192.168.74.251 \
 	--dst-db=db --dst-table=datatypes \
-	--mempool --mempool-max-events-num=3 --mempool-max-flush-interval=30 \
-	--csvpool --csvpool-file-path-prefix=qwe \
+	--csvpool --csvpool-file-path-prefix=qwe_ \
 	--csv-column-default-value date_1=2000-01-01 datetime_1=2000-01-01\ 01:02:03 time_1=2001-01-01\ 01:02:03 timestamp_1=2002-01-01\ 01:02:03
 
-#	--dst-file=dst.csv 
+#	--mempool --mempool-max-events-num=3 --mempool-max-flush-interval=30 \
+#	--dst-file=dst.csv
 #	--csvpool-keep-files

src/cliopts.py

@@ -74,7 +74,7 @@ def config():
         argparser.add_argument(
             '--mempool-max-events-num',
             type=int,
-            default=1000,
+            default=100000,
             help='max events num to pool before batch write'
         )
         argparser.add_argument(
@@ -91,7 +91,7 @@ def config():
         argparser.add_argument(
             '--csvpool-file-path-prefix',
             type=str,
-            default='/tmp/csvpool',
+            default='/tmp/csvpool_',
             help='file path prefix to CSV pool files'
         )
         argparser.add_argument(
@@ -220,7 +220,7 @@ def config():
                 'dry': args.dry,
                 'daemon': args.daemon,
                 'pid_file': args.pid_file,
-                'mempool': args.mempool,
+                'mempool': args.mempool or args.csvpool, # csvpool assumes mempool to be enabled
                 'mempool-max-events-num': args.mempool_max_events_num,
                 'mempool-max-flush-interval': args.mempool_max_flush_interval,
                 'csvpool': args.csvpool,

src/config.py

@@ -12,6 +12,7 @@
 from .objectbuilder import ObjectBuilder
 
 from .converter.csvwriteconverter import CSVWriteConverter
+from .converter.chwriteconverter import CHWriteConverter
 
 
 class Config(object):
@@ -42,6 +43,16 @@ def reader(self):
         else:
             return MySQLReader(**self.config['reader-config']['mysql'])
 
+    def converter_builder(self):
+        if not self.config['converter-config']['csv']['column_default_value']:
+            # no default values for CSV columns provided
+            return None
+
+        return ObjectBuilder(
+            instance=CSVWriteConverter(
+                defaults=self.config['converter-config']['csv']['column_default_value']
+            ))
+
     def writer_builder(self):
         if self.config['app-config']['csvpool']:
             return ObjectBuilder(class_name=ProcessWriter, constructor_params={
@@ -51,15 +62,21 @@ def writer_builder(self):
                         class_name=CHCSVWriter,
                         constructor_params=self.config['writer-config']['clickhouse']['connection_settings']
                     ),
-                    'converter_builder': ObjectBuilder(instance=CSVWriteConverter(defaults=self.config['converter-config']['csv']['column_default_value'])) if self.config['converter-config']['csv']['column_default_value'] else None,
+                    'converter_builder': self.converter_builder(),
                 })
             })
 
         elif self.config['writer-config']['file']['csv_file_path']:
-            return ObjectBuilder(class_name=CSVWriter, constructor_params=self.config['writer-config']['file'])
+            return ObjectBuilder(class_name=CSVWriter, constructor_params={
+                **self.config['writer-config']['file'],
+                'converter_builder': self.converter_builder(),
+            })
 
         else:
-            return ObjectBuilder(class_name=CHWriter, constructor_params=self.config['writer-config']['clickhouse'])
+            return ObjectBuilder(class_name=CHWriter, constructor_params={
+                **self.config['writer-config']['clickhouse'],
+                'converter_builder': ObjectBuilder(instance=CHWriteConverter()),
+            })
 
     def writer(self):
         if self.config['app-config']['mempool']:

src/writer/chwriter.py

@@ -3,8 +3,6 @@
 
 from clickhouse_driver.client import Client
 from .writer import Writer
-from ..event.event import Event
-from ..converter.chwriteconverter import CHWriteConverter
 
 
 class CHWriter(Writer):
@@ -13,7 +11,16 @@ class CHWriter(Writer):
     dst_db = None
     dst_table = None
 
-    def __init__(self, connection_settings, dst_db, dst_table):
+    def __init__(
+            self,
+            connection_settings,
+            dst_db=None,
+            dst_table=None,
+            next_writer_builder=None,
+            converter_builder=None,
+    ):
+        super().__init__(next_writer_builder=next_writer_builder, converter_builder=converter_builder)
+
         self.client = Client(**connection_settings)
         self.dst_db = dst_db
         self.dst_table = dst_table
@@ -32,16 +39,14 @@ def insert(self, event_or_events=None):
         if len(events) < 1:
             return
 
-        converter = CHWriteConverter()
-
         values = []
-        ev = None
+        event_converted = None
         for event in events:
-            ev = converter.convert(event)
-            values.append(ev.row)
+            event_converted = self.convert(event)
+            values.append(event_converted.row)
 
-        schema = self.dst_db if self.dst_db else ev.schema
-        table = self.dst_table if self.dst_table else ev.table
+        schema = self.dst_db if self.dst_db else event_converted.schema
+        table = self.dst_table if self.dst_table else event_converted.table
 
         try:
             sql = 'INSERT INTO `{0}`.`{1}` ({2}) VALUES'.format(

src/writer/csvwriter.py

@@ -40,7 +40,7 @@ def __init__(
         self.dst_table = dst_table
 
         if self.path is None:
-            self.path = self.path_prefix + '_' + '_'.join(self.path_suffix_parts) + '.csv'
+            self.path = self.path_prefix + '_'.join(self.path_suffix_parts) + '.csv'
             self.delete = not csv_keep_file
 
     def __del__(self):
@@ -82,7 +82,7 @@ def insert(self, event_or_events):
                 self.writer.writeheader()
 
         for event in events:
-            self.writer.writerow(self.converter_builder.get().convert(event).row if self.converter_builder else event.row)
+            self.writer.writerow(self.convert(event).row)
 
     def push(self):
         if not self.next_writer_builder:
@@ -93,7 +93,7 @@ def push(self):
         event.table = self.dst_table
         event.file = self.path
         event.fieldnames = self.fieldnames
-        self.next_writer_builder.get().insert([event])
+        self.next_writer_builder.get().insert(event)
 
     def close(self):
         if self.opened():

src/writer/writer.py

@@ -40,6 +40,8 @@ def listify(self, obj_or_list):
             # event_or_events is an object
             return [obj_or_list]
 
+    def convert(self, event):
+        return self.converter_builder.get().convert(event) if self.converter_builder else event
 
     def insert(self, event_or_events=None):
         # event_or_events = [