min and max rows per event

Author: sunsingerus

run_ontime.sh

@@ -2,7 +2,10 @@
 
 sudo bash -c "echo 1 > /proc/sys/net/ipv4/tcp_tw_reuse"
 
-python3.6 main.py ${*:1} \
+PYTHON=python3.6
+PYTHON=/home/user/pypy3.5-5.9-beta-linux_x86_64-portable/bin/pypy
+
+$PYTHON main.py ${*:1} \
     --src-resume \
     --src-wait \
     --nice-pause=1 \
@@ -13,7 +16,7 @@ python3.6 main.py ${*:1} \
     --csvpool \
     --csvpool-file-path-prefix=qwe_ \
     --mempool-max-flush-interval=60 \
-    --mempool-max-events-num=100000
+    --mempool-max-events-num=1000
 
 #	--mempool
 #   --mempool-max-events-num=3

src/reader/mysqlreader.py

@@ -69,17 +69,21 @@ def __init__(
             resume_stream=self.resume_stream,
         )
 
-    def speed_report(self, start, rows_num, now=None):
+    def performance_report(self, start, rows_num, rows_per_event_min=-1, rows_per_event_max=-1, now=None):
         # time to calc stat
+
         if now is None:
             now = time.time()
+
         window_size = now - start
         rows_per_sec = rows_num / window_size
         logging.info(
-            'rows_per_sec:%f for last %d rows %f sec',
+            'rows_per_sec:%f rows_per_event_min: %d rows_per_event_max: %d for last %d rows %f sec',
             rows_per_sec,
+            rows_per_event_min,
+            rows_per_event_max,
             rows_num,
-            window_size
+            window_size,
         )
 
     def read(self):
@@ -91,16 +95,27 @@ def read(self):
 
                 start = time.time()
                 rows_num = 0
-                rows_num_since_interim_speed_report = 0
+                rows_num_since_interim_performance_report = 0
+                rows_per_event = 0
+                rows_per_event_min = 0
+                rows_per_event_max = 0
+
 
                 # fetch available events from MySQL
                 for mysql_event in self.binlog_stream:
                     if isinstance(mysql_event, WriteRowsEvent):
+
+                        rows_per_event = len(mysql_event.rows)
+                        if rows_per_event < rows_per_event_min:
+                            rows_per_event_min = rows_per_event
+                        if rows_per_event > rows_per_event_max:
+                            rows_per_event_max = rows_per_event
+
                         if self.subscribers('WriteRowsEvent'):
                             self.write_rows_event_num += 1
                             logging.debug('WriteRowsEvent #%d rows: %d', self.write_rows_event_num, len(mysql_event.rows))
                             rows_num += len(mysql_event.rows)
-                            rows_num_since_interim_speed_report += len(mysql_event.rows)
+                            rows_num_since_interim_performance_report += len(mysql_event.rows)
                             event = Event()
                             event.schema = mysql_event.schema
                             event.table = mysql_event.table
@@ -112,17 +127,24 @@ def read(self):
                             logging.debug('WriteRowsEvent.EachRow #%d', self.write_rows_event_each_row_num)
                             for row in mysql_event.rows:
                                 rows_num += 1
-                                rows_num_since_interim_speed_report += 1
+                                rows_num_since_interim_performance_report += 1
                                 event = Event()
                                 event.schema = mysql_event.schema
                                 event.table = mysql_event.table
                                 event.row = row['values']
                                 self.notify('WriteRowsEvent.EachRow', event=event)
 
-                        if rows_num_since_interim_speed_report >= 100000:
+                        if rows_num_since_interim_performance_report >= 100000:
                             # speed report each N rows
-                            self.speed_report(start, rows_num)
-                            rows_num_since_interim_speed_report = 0
+                            self.performance_report(
+                                start=start,
+                                rows_num=rows_num,
+                                rows_per_event_min=rows_per_event_min,
+                                rows_per_event_max=rows_per_event_max,
+                            )
+                            rows_num_since_interim_performance_report = 0
+                            rows_per_event_min = 0
+                            rows_per_event_max = 0
                     else:
                         # skip non-insert events
                         pass
@@ -134,7 +156,7 @@ def read(self):
                     now = time.time()
                     if now > start + 60:
                         # and processing was long enough
-                        self.speed_report(start, rows_num, now)
+                        self.performance_report(start, rows_num, now)
 
                 if not self.blocking:
                     break # while True