pass schema.table info through csv pool

Author: sunsingerus

run.sh

@@ -1,13 +1,22 @@
 #!/bin/bash
 
 python3 main.py \
-	--src-resume --src-wait \
-	--src-host=127.0.0.1 --src-user=reader --src-password=qwerty \
-	--dst-host=192.168.74.251 \
-	--dst-db=db --dst-table=datatypes \
-	--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
+    --src-resume \
+    --src-wait \
+    --src-host=127.0.0.1 \
+    --src-user=reader \
+    --src-password=qwerty \
+    --dst-host=192.168.74.251 \
+    --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 \
+    --mempool-max-flush-interval=600 \
+    --mempool-max-events-num=900000
 
-#	--mempool --mempool-max-events-num=3 --mempool-max-flush-interval=30 \
+#	--mempool
+#   --mempool-max-events-num=3
+#   --mempool-max-flush-interval=30
 #	--dst-file=dst.csv
+#	--dst-schema=db
+#   --dst-table=datatypes
 #	--csvpool-keep-files

src/cliopts.py

@@ -52,7 +52,7 @@ def config():
         argparser.add_argument(
             '--dry',
             action='store_true',
-            help='Dry mode - do not do anything that can harm.'
+            help='Dry mode - do not do anything that can harm. '
             'Useful for debugging.'
         )
         argparser.add_argument(
@@ -75,29 +75,29 @@ def config():
             '--mempool-max-events-num',
             type=int,
             default=100000,
-            help='max events num to pool before batch write'
+            help='Max events number to pool - triggering pool flush'
         )
         argparser.add_argument(
             '--mempool-max-flush-interval',
             type=int,
             default=60,
-            help='max seconds num between flushes'
+            help='Max seconds number between pool flushes'
         )
         argparser.add_argument(
             '--csvpool',
             action='store_true',
-            help='Cache data in csv files.'
+            help='Cache data in CSV pool files on disk. Requires memory pooling, thus enables --mempool even if it is not explicitly specified'
         )
         argparser.add_argument(
             '--csvpool-file-path-prefix',
             type=str,
             default='/tmp/csvpool_',
-            help='file path prefix to CSV pool files'
+            help='File path prefix to CSV pool files'
         )
         argparser.add_argument(
             '--csvpool-keep-files',
             action='store_true',
-            help='Keep pool csv files.'
+            help='Keep CSV pool files. Useful for debugging'
         )
 
         argparser.add_argument(
@@ -156,7 +156,7 @@ def config():
             '--src-file',
             type=str,
             default=None,
-            help='Source file tp read data from'
+            help='Source file to read data from'
         )
 
         argparser.add_argument(
@@ -190,10 +190,10 @@ def config():
             help='Password to be used when writing to dst'
         )
         argparser.add_argument(
-            '--dst-db',
+            '--dst-schema',
             type=str,
             default=None,
-            help='Database to be used when writing to dst'
+            help='Database/schema to be used when writing to dst'
         )
         argparser.add_argument(
             '--dst-table',
@@ -262,15 +262,15 @@ def config():
                         'user': args.dst_user,
                         'password': args.dst_password,
                     },
-                    'dst_db': args.dst_db,
+                    'dst_schema': args.dst_schema,
                     'dst_table': args.dst_table,
                 },
                 'file': {
                     'csv_file_path': args.dst_file,
                     'csv_file_path_prefix': args.csvpool_file_path_prefix,
                     'csv_file_path_suffix_parts': [],
                     'csv_keep_file': args.csvpool_keep_files,
-                    'dst_db': args.dst_db,
+                    'dst_schema': args.dst_schema,
                     'dst_table': args.dst_table,
                 },
             },

src/pool/bbpool.py

@@ -89,17 +89,17 @@ def flush(self, key=None):
     def rotate_belt(self, belt_index, flush=False):
         now = int(time.time())
         need_rotation = True if flush else False
-        rotate_by = "FLUSH"
+        rotate_reason = "FLUSH"
 
         if len(self.belts[belt_index][0]) >= self.max_bucket_size:
             # 0-index bucket is full
             need_rotation = True
-            rotate_by = "SIZE"
+            rotate_reason = "SIZE"
 
         elif now >= self.belts_rotated_at[belt_index] + self.max_interval_between_rotations:
             # time interval reached
             need_rotation = True
-            rotate_by = "TIME"
+            rotate_reason = "TIME"
 
         if not need_rotation:
             # belt not rotated
@@ -121,7 +121,7 @@ def rotate_belt(self, belt_index, flush=False):
 
             buckets_num = len(self.belts[belt_index])
             last_bucket_size = len(self.belts[belt_index][buckets_num-1])
-            print(now, self.buckets_count, 'rotating belt', belt_index, 'rotate by', rotate_by, 'buckets_num', buckets_num, 'last bucket size', last_bucket_size, 'belts:', len(self.belts))
+            print('rotating belt. now:', now, 'bucket number:', self.buckets_count, 'index:', belt_index, 'reason:', rotate_reason, 'buckets on belt:', buckets_num, 'last bucket size:', last_bucket_size, 'belts count:', len(self.belts))
 
             # time to flush data for specified key
             self.writer_builder.param('csv_file_path_suffix_parts', [str(now), str(self.buckets_count)])

src/writer/chcsvwriter.py

@@ -55,7 +55,7 @@ def insert(self, event_or_events=None):
                 sql,
             )
 
-            print('running:', bash)
+#            print('running:', bash)
             os.system(bash)
 
         pass

src/writer/chwriter.py

@@ -8,21 +8,21 @@
 class CHWriter(Writer):
 
     client = None
-    dst_db = None
+    dst_schema = None
     dst_table = None
 
     def __init__(
             self,
             connection_settings,
-            dst_db=None,
+            dst_schema=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_schema = dst_schema
         self.dst_table = dst_table
 
     def insert(self, event_or_events=None):
@@ -45,7 +45,7 @@ def insert(self, event_or_events=None):
             event_converted = self.convert(event)
             values.append(event_converted.row)
 
-        schema = self.dst_db if self.dst_db else event_converted.schema
+        schema = self.dst_schema if self.dst_schema else event_converted.schema
         table = self.dst_table if self.dst_table else event_converted.table
 
         try:

src/writer/csvwriter.py

@@ -12,7 +12,7 @@ class CSVWriter(Writer):
     file = None
     path = None
     writer = None
-    dst_db = None
+    dst_schema = None
     dst_table = None
     fieldnames = None
     header_written = False
@@ -26,7 +26,7 @@ def __init__(
             csv_file_path_prefix=None,
             csv_file_path_suffix_parts=[],
             csv_keep_file=False,
-            dst_db=None,
+            dst_schema=None,
             dst_table=None,
             next_writer_builder=None,
             converter_builder=None,
@@ -36,7 +36,7 @@ def __init__(
         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_schema = dst_schema
         self.dst_table = dst_table
 
         if self.path is None:
@@ -77,6 +77,11 @@ def insert(self, event_or_events):
 
         if not self.writer:
             self.fieldnames = sorted(events[0].row.keys())
+            if self.dst_schema is None:
+                self.dst_schema = events[0].schema
+            if self.dst_table is None:
+                self.dst_table = events[0].table
+
             self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames)
             if not self.header_written:
                 self.writer.writeheader()
@@ -89,7 +94,7 @@ def push(self):
             return
 
         event = Event()
-        event.schema = self.dst_db
+        event.schema = self.dst_schema
         event.table = self.dst_table
         event.file = self.path
         event.fieldnames = self.fieldnames

src/writer/processwriter.py

@@ -39,7 +39,7 @@ def insert(self, event_or_events=None):
         #   },
         # ]
         process = mp.Process(target=self.process, args=(event_or_events,))
-        print('Start Process')
+        #print('Start Process')
         process.start()
         #print('Join Process')
         #process.join()