Merge pull request #21 from tinybirdco/unify-syncers

unify syncers in one datasource

Author: ricardotinybird

clickhouse_mysql/pool/bbpool.py

@@ -149,18 +149,7 @@ def rotate_belt(self, belt_index, flush=False):
             # time to flush data for specified key
             #self.writer_builder.param('csv_file_path_suffix_parts', [str(int(now)), str(self.buckets_num_total)])
             writer = self.writer_builder.new()
-            item = self.belts[belt_index].pop()
-            # process event based on its type
-            if isinstance(item[0].pymysqlreplication_event, WriteRowsEvent):
-                writer.insert(item)
-            elif isinstance(item[0].pymysqlreplication_event, DeleteRowsEvent):
-                writer.delete(item)
-            elif isinstance(item[0].pymysqlreplication_event, UpdateRowsEvent):
-                writer.update(item)
-            else:
-                # skip other unhandled events
-                pass
-            # writer.insert(self.belts[belt_index].pop())
+            writer.insert(self.belts[belt_index].pop())
             writer.close()
             writer.push()
             writer.destroy()

clickhouse_mysql/reader/mysqlreader.py

@@ -239,7 +239,7 @@ def stat_write_rows_event_finalyse(self):
 
     def process_first_event(self, event):
         if "{}.{}".format(event.schema, event.table) not in self.first_rows_passed:
-            Util.log_row(event.first_row(), "first row in replication {}.{}".format(event.schema, event.table))
+            Util.log_row(event.first_row(), "first row in replication {}.{} - binlog pos {}".format(event.schema, event.table, event.pymysqlreplication_event.packet.log_pos))
             self.first_rows_passed.append("{}.{}".format(event.schema, event.table))
         logging.info(self.first_rows_passed)
 
@@ -403,8 +403,7 @@ def read(self):
                 self.stat_init_fetch_loop()
 
                 try:
-                    logging.debug('Pre-start binlog position: ' + self.binlog_stream.log_file + ":" + str(
-                        self.binlog_stream.log_pos) if self.binlog_stream.log_pos is not None else "undef")
+                    logging.debug('Pre-start binlog position: ' + self.binlog_stream.log_file + ":" + str(self.binlog_stream.log_pos) if self.binlog_stream.log_pos is not None else "undef")
 
                     # fetch available events from MySQL
                     for mysql_event in self.binlog_stream:
@@ -426,8 +425,8 @@ def read(self):
                             # skip other unhandled events
                             pass
 
-                    # after event processed, we need to handle current binlog position
-                    self.process_binlog_position(self.binlog_stream.log_file, self.binlog_stream.log_pos)
+                        # after event processed, we need to handle current binlog position
+                        self.process_binlog_position(self.binlog_stream.log_file, self.binlog_stream.log_pos)
 
                 except Exception as ex:
                     if self.blocking:

clickhouse_mysql/writer/csvwriter.py

@@ -7,6 +7,7 @@
 import copy
 import time
 import uuid
+import json
 
 from clickhouse_mysql.writer.writer import Writer
 from clickhouse_mysql.event.event import Event
@@ -125,9 +126,17 @@ def insert(self, event_or_events):
 
             event_converted = self.convert(event)
             rows = event_converted.pymysqlreplication_event.rows
-            headers = list(rows[0]['values'].keys())
+            if 'after_values' in rows[0].keys():
+                headers = list(rows[0]['after_values'].keys())
+            else:
+                headers = list(rows[0]['values'].keys())
             headers.insert(0, 'operation')
             headers.insert(1, 'tb_upd')
+            headers.insert(2, 'table')
+            headers.insert(3, 'schema')
+            headers.insert(4, 'log_pos')
+            headers.insert(5, 'binlog_timestamp')
+            headers.insert(6, 'payload')
 
             # self.fieldnames = sorted(self.convert(copy.copy(event.first_row())).keys())
             self.fieldnames = headers
@@ -136,59 +145,7 @@ def insert(self, event_or_events):
             if self.dst_table is None:
                 self.dst_table = event.table
 
-            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_NONNUMERIC)
-            if not self.header_written:
-                self.writer.writeheader()
-
-        for event in events:
-            if not event.verify:
-                logging.warning('Event verification failed. Skip one event. Event: %s Class: %s', event.meta(), __class__)
-                continue # for event
-            self.generate_row(event)
-
-    def delete_row(self, event_or_events):
-
-        # event_or_events = [
-        #   event: {
-        #       row: {'id': 3, 'a': 3}
-        #   },
-        #   event: {
-        #       row: {'id': 3, 'a': 3}
-        #   },
-        # ]
-
-        logging.debug("Delete CSV Writer")
-
-        events = self.listify(event_or_events)
-        if len(events) < 1:
-            logging.warning('No events to delete. class: %s', __class__)
-            return
-
-        # assume we have at least one Event
-
-        logging.debug('class:%s delete %d events', __class__, len(events))
-
-        if not self.opened():
-            self.open()
-
-        if not self.writer:
-            # pick any event from the list
-            event = events[0]
-            if not event.verify:
-                logging.warning('Event verification failed. Skip insert(). Event: %s Class: %s', event.meta(), __class__)
-                return
-
-            event_converted = self.convert(event)
-            rows = event_converted.pymysqlreplication_event.rows
-            headers = list(rows[0]['values'].keys())
-            headers.insert(0, 'operation')
-            headers.insert(1, 'tb_upd')
-            
-            self.fieldnames = headers
-            if self.dst_schema is None:
-                self.dst_schema = event.schema
-            if self.dst_table is None:
-                self.dst_table = event.table
+            self.fieldnames = self.fieldnames[0:7]  # get only operation, tb_upd, table and payload
 
             self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_NONNUMERIC)
             if not self.header_written:
@@ -200,72 +157,6 @@ def delete_row(self, event_or_events):
                 continue # for event
             self.generate_row(event)
 
-
-
-    def update(self, event_or_events):
-        
-        # event_or_events = [
-        #   event: {
-        #       row: {
-        #           'before_values': {'id': 3, 'a': 3},
-        #           'after_values': {'id': 3, 'a': 2}
-        #       }
-        #   },
-        #   event: {
-        #       row: {
-        #          'before_values': {'id': 2, 'a': 3},
-        #          'after_values': {'id': 2, 'a': 2}
-        #       }
-        #   },
-        # ]
-
-        logging.debug("Update CSV Writer")
-
-        events = self.listify(event_or_events)
-        if len(events) < 1:
-            logging.warning('No events to update. class: %s', __class__)
-            return
-
-        # assume we have at least one Event
-
-        logging.debug('class:%s updated %d events', __class__, len(events))
-
-        if not self.opened():
-            self.open()
-
-        if not self.writer:
-            # pick any event from the list
-            event = events[0]
-            if not event.verify:
-                logging.warning('Event verification failed. Skip insert(). Event: %s Class: %s', event.meta(), __class__)
-                return
-
-            event_converted = self.convert(event)
-            rows = event_converted.pymysqlreplication_event.rows
-            headers = list(rows[0]['after_values'].keys())
-            headers.insert(0, 'operation')
-            headers.insert(1, 'tb_upd')
-            
-            # self.fieldnames = sorted(headers)
-            self.fieldnames = headers
-            if self.dst_schema is None:
-                self.dst_schema = event.schema
-            if self.dst_table is None:
-                self.dst_table = event.table
-
-            self.writer = csv.DictWriter(self.file, fieldnames=self.fieldnames, quoting=csv.QUOTE_NONNUMERIC)
-            if not self.header_written:
-                self.writer.writeheader()
-
-        for event in events:
-            if not event.verify:
-                logging.warning('Event verification failed. Skip one event. Event: %s Class: %s', event.meta(), __class__)
-                continue # for event
-            
-            event_converted = self.convert(event)
-            self.generate_row(event_converted)
-
-
     def convert_null_values(self, row):
         """ We need to mark those fields that are null to be able to distinguish between NULL and empty strings """
         for key in list(row.keys()):
@@ -274,26 +165,22 @@ def convert_null_values(self, row):
 
     def generate_row(self, event):
         """ When using mempool or csvpool events are cached so you can receive different kind of events in the same list. These events should be handled in a different way """
-
-        if isinstance(event.pymysqlreplication_event, WriteRowsEvent):
-            for row in event:
-                row['tb_upd'] = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
-                row['operation'] = 0
-                self.convert_null_values(row)
-                self.writer.writerow(self.convert(row))
-        elif isinstance(event.pymysqlreplication_event, DeleteRowsEvent):
-            for row in event:
-                row['tb_upd'] = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
-                row['operation'] = 2
-                self.convert_null_values(row)
-                self.writer.writerow(self.convert(row))
-        elif isinstance(event.pymysqlreplication_event, UpdateRowsEvent):
-            for row in event.pymysqlreplication_event.rows:
-                row['after_values']['tb_upd'] = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
-                row['after_values']['operation'] = 1
-                self.convert_null_values(row['after_values'])
-                self.writer.writerow(self.convert(row['after_values']))
-    
+        row_w_payload = {}
+        for row in event:
+            row_w_payload['tb_upd'] = datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S.%f')
+            if isinstance(event.pymysqlreplication_event, WriteRowsEvent):
+                row_w_payload['operation'] = 0
+            elif isinstance(event.pymysqlreplication_event, DeleteRowsEvent):
+                row_w_payload['operation'] = 2
+            else:
+                row_w_payload['operation'] = 1
+            row_w_payload['table'] = event.table
+            row_w_payload['schema'] = str(event.schema).split('_')[0]
+            row_w_payload['log_pos'] = event.pymysqlreplication_event.packet.log_pos
+            row_w_payload['binlog_timestamp'] = event.pymysqlreplication_event.timestamp
+            self.convert_null_values(row)
+            row_w_payload['payload'] = json.dumps(row, default=str)
+            self.writer.writerow(self.convert(row_w_payload))
 
     def push(self):
         if not self.next_writer_builder or not self.fieldnames:

clickhouse_mysql/writer/processwriter.py

@@ -6,7 +6,6 @@
 
 from clickhouse_mysql.writer.writer import Writer
 
-
 class ProcessWriter(Writer):
     """Start write procedure as a separated process"""
     args = None
@@ -35,28 +34,6 @@ def process(self, event_or_events=None):
         writer.destroy()
         logging.debug('class:%s process() done', __class__)
 
-    def processDelete(self, event_or_events=None):
-        """Separate process body to be run"""
-
-        logging.debug('class:%s process()', __class__)
-        writer = self.next_writer_builder.get()
-        writer.delete_row(event_or_events)
-        writer.close()
-        writer.push()
-        writer.destroy()
-        logging.debug('class:%s processDelete() done', __class__)
-
-    def processUpdate(self, event_or_events=None):
-        """Separate process body to be run"""
-
-        logging.debug('class:%s process()', __class__)
-        writer = self.next_writer_builder.get()
-        writer.update(event_or_events)
-        writer.close()
-        writer.push()
-        writer.destroy()
-        logging.debug('class:%s processUpdate() done', __class__)
-
     def insert(self, event_or_events=None):
         # event_or_events = [
         #   event: {
@@ -79,50 +56,6 @@ def insert(self, event_or_events=None):
         logging.debug('class:%s insert done', __class__)
         pass
 
-    def delete(self, event_or_events=None):
-        # event_or_events = [
-        #   event: {
-        #       row: {'id': 3, 'a': 3}
-        #   },
-        #   event: {
-        #       row: {'id': 3, 'a': 3}
-        #   },
-        # ]
-
-        # start separated process with event_or_events to be inserted
-
-        logging.debug('class:%s delete', __class__)
-        process = mp.Process(target=self.processDelete, args=(event_or_events,))
-
-        logging.debug('class:%s delete.process.start()', __class__)
-        process.start()
-
-        #process.join()
-        logging.debug('class:%s delete done', __class__)
-        pass
-
-    def update(self, event_or_events=None):
-        # event_or_events = [
-        #   event: {
-        #       row: {'id': 3, 'a': 3}
-        #   },
-        #   event: {
-        #       row: {'id': 3, 'a': 3}
-        #   },
-        # ]
-
-        # start separated process with event_or_events to be inserted
-
-        logging.debug('class:%s update', __class__)
-        process = mp.Process(target=self.processUpdate, args=(event_or_events,))
-
-        logging.debug('class:%s update.process.start()', __class__)
-        process.start()
-
-        #process.join()
-        logging.debug('class:%s update done', __class__)
-        pass
-
     def flush(self):
         pass
 

clickhouse_mysql/writer/tbcsvwriter.py

@@ -52,30 +52,15 @@ def __init__(
         self.dst_distribute = dst_distribute
 
 
-    def format_null_values(self, csv_file):
-        """ We need to replace NULL values by \\N (CH null value) and do not quote this field. 
-        It is the only way to distinguish between NULL and empty strings. With this we will have: 
-          - xx,\\N,yy --> For null values
-          - xx,'',yy --> For empty strings  """
-        
-        with open(csv_file, 'r') as file:
-            data = file.read()
-            data = data.replace('"NULL"', '\\N')    
-        
-        with open(csv_file, 'w') as file:
-            file.write(data)
-
-
     def uploadCSV(self, table, filename, tries=1):
         limit_of_retries = 3
         params = {
             'name': table,
-            'mode': 'append'
+            'mode': 'append',
+            'dialect_delimiter': ','
         }
 
         try:
-            # Add replace NULL values by \N
-            self.format_null_values(filename)
 
             with open(filename, 'rb') as f:
                 m = MultipartEncoder(fields={'csv': ('csv', f, 'text/csv')})
@@ -264,6 +249,6 @@ def update(self, event_or_events=None):
         #     logging.debug('starting %s', bash)
         #     os.system(bash)
 
-        logging.debug("CHCSVWriter: delete row")
+        logging.debug("CHCSVWriter: update row")
 
         pass

pack/clickhouse-mysql.spec

@@ -118,7 +118,7 @@ chown -R $CLICKHOUSE_USER:$CLICKHOUSE_GROUP /var/run/clickhouse-mysql
 chown -R $CLICKHOUSE_USER:$CLICKHOUSE_GROUP /var/log/clickhouse-mysql
 
 /usr/bin/pip3 install mysqlclient
-/usr/bin/pip3 install mysql-replication
+/usr/bin/pip3 install mysql-replication==0.45.1
 /usr/bin/pip3 install clickhouse-driver
 /usr/bin/pip3 install configobj
 

requirements.txt

@@ -1,5 +1,5 @@
 clickhouse-driver==0.2.0
-mysql-replication==0.23
+mysql-replication==0.45.1
 mysqlclient==2.0.3
 PyMySQL==1.0.2
 pytz==2021.1