process writer

Author: sunsingerus

main.py

@@ -6,6 +6,7 @@
 from src.daemon import Daemon
 
 import sys
+import multiprocessing as mp
 
 
 if sys.version_info[0] < 3:
@@ -17,6 +18,7 @@ class Main(Daemon):
     config = None
 
     def __init__(self):
+        mp.set_start_method('forkserver')
         self.config = CLIOpts.config()
         super().__init__(pidfile=self.config.pid_file())
 

src/config.py

@@ -8,6 +8,7 @@
 from .writer.csvwriter import CSVWriter
 from .writer.chcsvwriter import CHCSVWriter
 from .writer.poolwriter import PoolWriter
+from .writer.processwriter import ProcessWriter
 from .objectbuilder import ObjectBuilder
 
 from .converter.csvwriteconverter import CSVWriteConverter
@@ -42,19 +43,23 @@ def reader(self):
             return MySQLReader(**self.config['reader-config']['mysql'])
 
     def writer_builder(self):
-
         if self.config['app-config']['csvpool']:
-            return ObjectBuilder(class_name=CSVWriter, params={
-                **self.config['writer-config']['file'],
-                'next_writer_builder': ObjectBuilder(instance=CHCSVWriter(**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,
+            return ObjectBuilder(class_name=ProcessWriter, constructor_params={
+                'next_writer_builder': ObjectBuilder(class_name=CSVWriter, constructor_params={
+                    **self.config['writer-config']['file'],
+                    'next_writer_builder': ObjectBuilder(
+                        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,
+                })
             })
 
         elif self.config['writer-config']['file']['csv_file_path']:
-            return ObjectBuilder(class_name=CSVWriter, params=self.config['writer-config']['file'])
+            return ObjectBuilder(class_name=CSVWriter, constructor_params=self.config['writer-config']['file'])
 
         else:
-            return ObjectBuilder(class_name=CHWriter, params=self.config['writer-config']['clickhouse'])
+            return ObjectBuilder(class_name=CHWriter, constructor_params=self.config['writer-config']['clickhouse'])
 
     def writer(self):
         if self.config['app-config']['mempool']:

src/objectbuilder.py

@@ -5,26 +5,27 @@
 class ObjectBuilder(object):
 
     class_name = None
-    params = None
+    constructor_params = None
     instance = None
 
-    def __init__(self, class_name=None, params=None, instance=None):
+    def __init__(self, class_name=None, constructor_params=None, instance=None):
         self.class_name = class_name
-        self.params = params
+        self.constructor_params = constructor_params
         self.instance = instance
 
     def param(self, name, value):
-        self.params[name] = value
+        if not self.constructor_params:
+            self.constructor_params = {}
+        self.constructor_params[name] = value
 
     def get(self):
-
         if not self.class_name:
             # no class name - return instance, it may be None
             return self.instance
 
         # have class name
 
-        if self.params:
-            return self.class_name(**self.params)
+        if self.constructor_params:
+            return self.class_name(**self.constructor_params)
         else:
             return self.class_name()

src/pumper.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
+import time
 
 class Pumper(object):
 

src/reader/mysqlreader.py

@@ -88,7 +88,6 @@ def read(self):
 
                 # blocking
                 self.fire('ReaderIdleEvent')
-                time.sleep(1)
 
         except KeyboardInterrupt:
             pass

src/writer/chcsvwriter.py

@@ -1,9 +1,12 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
+from .writer import Writer
+
 import os
+import time
 
-class CHCSVWriter(object):
+class CHCSVWriter(Writer):
 
     host = None
     port = None
@@ -26,7 +29,11 @@ def insert(self, event_or_events=None):
         #   },
         # ]
 
-        for event in event_or_events:
+        events = self.listify(event_or_events)
+        if len(events) < 1:
+            return
+
+        for event in events:
             sql = 'INSERT INTO `{0}`.`{1}` ({2}) FORMAT CSV'.format(
                 event.schema,
                 event.table,

src/writer/chwriter.py

@@ -28,24 +28,15 @@ def insert(self, event_or_events=None):
         #   },
         # ]
 
-        if event_or_events is None:
-            # nothing to insert at all
+        events = self.listify(event_or_events)
+        if len(events) < 1:
             return
 
-        elif isinstance(event_or_events, list):
-            if len(event_or_events) < 1:
-                # list is empty - nothing to insert
-                return
-
-        else:
-            # event_or_events is instance of Event
-            event_or_events = [event_or_events]
-
         converter = CHWriteConverter()
 
         values = []
         ev = None
-        for event in event_or_events:
+        for event in events:
             ev = converter.convert(event)
             values.append(ev.row)
 

src/writer/csvwriter.py

@@ -31,13 +31,13 @@ def __init__(
             next_writer_builder=None,
             converter_builder=None,
     ):
+        super().__init__(next_writer_builder=next_writer_builder, converter_builder=converter_builder)
+
         self.path = csv_file_path
         self.path_prefix = csv_file_path_prefix
         self.path_suffix_parts = csv_file_path_suffix_parts
         self.dst_db = dst_db
         self.dst_table = dst_table
-        self.next_writer_builder = next_writer_builder
-        self.converter_builder = converter_builder
 
         if self.path is None:
             self.path = self.path_prefix + '_' + '_'.join(self.path_suffix_parts) + '.csv'
@@ -68,29 +68,20 @@ def insert(self, event_or_events):
         #   },
         # ]
 
-        if event_or_events is None:
-            # nothing to insert at all
+        events = self.listify(event_or_events)
+        if len(events) < 1:
             return
 
-        elif isinstance(event_or_events, list):
-            if len(event_or_events) < 1:
-                # list is empty - nothing to insert
-                return
-
-        else:
-            # event_or_events is instance of Event
-            event_or_events = [event_or_events]
-
         if not self.opened():
             self.open()
 
         if not self.writer:
-            self.fieldnames = sorted(event_or_events[0].row.keys())
+            self.fieldnames = sorted(events[0].row.keys())
             self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames)
             if not self.header_written:
                 self.writer.writeheader()
 
-        for event in event_or_events:
+        for event in events:
             self.writer.writerow(self.converter_builder.get().convert(event).row if self.converter_builder else event.row)
 
     def push(self):

src/writer/processwriter.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from .writer import Writer
+import multiprocessing as mp
+
+
+class ProcessWriter(Writer):
+
+    args = None
+
+    def __init__(self, **kwargs):
+        next_writer_builder = kwargs.pop('next_writer_builder', None)
+        converter_builder = kwargs.pop('converter_builder', None)
+        super().__init__(next_writer_builder=next_writer_builder, converter_builder=converter_builder)
+        for arg in kwargs:
+            self.next_writer_builder.param(arg, kwargs[arg])
+
+    def opened(self):
+        pass
+
+    def open(self):
+        pass
+
+    def process(self, event_or_events=None):
+        writer = self.next_writer_builder.get()
+        writer.insert(event_or_events)
+        writer.close()
+        writer.push()
+        writer.destroy()
+
+    def insert(self, event_or_events=None):
+        # event_or_events = [
+        #   event: {
+        #       row: {'id': 3, 'a': 3}
+        #   },
+        #   event: {
+        #       row: {'id': 3, 'a': 3}
+        #   },
+        # ]
+        process = mp.Process(target=self.process, args=(event_or_events,))
+        print('Start Process')
+        process.start()
+        #print('Join Process')
+        #process.join()
+        #print('Done Process')
+        pass
+
+    def flush(self):
+        pass
+
+    def push(self):
+        pass
+
+    def destroy(self):
+        pass
+
+    def close(self):
+        pass

src/writer/writer.py

@@ -7,12 +7,40 @@ class Writer(object):
     next_writer_builder = None
     converter_builder = None
 
+    def __init__(
+            self,
+            next_writer_builder=None,
+            converter_builder=None
+    ):
+        self.next_writer_builder = next_writer_builder
+        self.converter_builder = converter_builder
+
     def opened(self):
         pass
 
     def open(self):
         pass
 
+    def listify(self, obj_or_list):
+        """Ensure list"""
+
+        if obj_or_list is None:
+            # no value - return empty list
+            return []
+
+        elif isinstance(obj_or_list, list):
+            if len(obj_or_list) < 1:
+                # list is empty - nothing to do
+                return []
+            else:
+                # list is good
+                return obj_or_list
+
+        else:
+            # event_or_events is an object
+            return [obj_or_list]
+
+
     def insert(self, event_or_events=None):
         # event_or_events = [
         #   event: {