Migration Strategies

-- 1. Design Cassandra data model -- Map MySQL schema to Cassandra query patterns -- MySQL: users table CREATE TABLE users ( id INT PRIMARY KEY, email VARCHAR(255) UNIQUE, username VARCHAR(100), created_at TIMESTAMP ); -- Cassandra: users_by_id (primary lookup) CREATE TABLE users_by_id ( user_id UUID PRIMARY KEY, email text, username text, created_at timestamp ); -- Cassandra: users_by_email (lookup by email) CREATE TABLE users_by_email ( email text PRIMARY KEY, user_id UUID, username text, created_at timestamp ); -- 2. Set up Cassandra cluster -- 3. Create tables and indexes -- 4. Test connectivity from application

# Python script: Migrate existing MySQL data to Cassandra import mysql.connector from cassandra.cluster import Cluster import uuid # Connect to both databases mysql_conn = mysql.connector.connect( host='localhost', user='root', database='myapp' ) cassandra_cluster = Cluster(['127.0.0.1']) cassandra_session = cassandra_cluster.connect('myapp') # Prepared statements for Cassandra insert_by_id = cassandra_session.prepare(""" INSERT INTO users_by_id (user_id, email, username, created_at) VALUES (?, ?, ?, ?) """) insert_by_email = cassandra_session.prepare(""" INSERT INTO users_by_email (email, user_id, username, created_at) VALUES (?, ?, ?, ?) """) # Batch migrate users (1000 at a time) mysql_cursor = mysql_conn.cursor() offset = 0 batch_size = 1000 while True: mysql_cursor.execute(""" SELECT id, email, username, created_at FROM users LIMIT %s OFFSET %s """, (batch_size, offset)) rows = mysql_cursor.fetchall() if not rows: break for row in rows: user_id = uuid.uuid4() # Generate new UUID # Write to both Cassandra tables cassandra_session.execute(insert_by_id, (user_id, row[1], row[2], row[3])) cassandra_session.execute(insert_by_email, (row[1], user_id, row[2], row[3])) offset += batch_size print(f"Migrated {offset} users...") print("✅ Historical data migration complete!") # ⚠️ Run during off-peak hours! # ⚠️ Monitor MySQL and Cassandra load

# Application code: Write to both databases class UserRepository: def __init__(self, mysql_conn, cassandra_session): self.mysql = mysql_conn self.cassandra = cassandra_session self.dual_write_enabled = True # Feature flag def create_user(self, email, username): user_id = uuid.uuid4() created_at = datetime.now() try: # 1. Write to MySQL (source of truth during migration) mysql_cursor = self.mysql.cursor() mysql_cursor.execute(""" INSERT INTO users (email, username, created_at) VALUES (%s, %s, %s) """, (email, username, created_at)) mysql_id = mysql_cursor.lastrowid self.mysql.commit() # 2. Write to Cassandra (if dual write enabled) if self.dual_write_enabled: try: self.cassandra.execute(""" INSERT INTO users_by_id (user_id, email, username, created_at) VALUES (%s, %s, %s, %s) """, (user_id, email, username, created_at)) self.cassandra.execute(""" INSERT INTO users_by_email (email, user_id, username, created_at) VALUES (%s, %s, %s, %s) """, (email, user_id, username, created_at)) except Exception as e: # Log error but DON'T fail request # MySQL write succeeded, so user is created logger.error(f"Cassandra write failed: {e}") # Alert ops team for investigation return {'id': mysql_id, 'email': email} except Exception as e: self.mysql.rollback() raise e # ✅ Key points: # - MySQL = source of truth (fails request if write fails) # - Cassandra = best effort (log errors, don't fail) # - Feature flag allows quick disable if issues

# Validate data consistency by reading from both def get_user_by_email(self, email): # 1. Read from MySQL (still serving production traffic) mysql_cursor = self.mysql.cursor() mysql_cursor.execute(""" SELECT id, email, username, created_at FROM users WHERE email = %s """, (email,)) mysql_result = mysql_cursor.fetchone() # 2. Shadow read from Cassandra (for validation) if self.shadow_read_enabled: try: cassandra_result = self.cassandra.execute(""" SELECT user_id, email, username, created_at FROM users_by_email WHERE email = %s """, (email,)).one() # Compare results (asynchronously) self.compare_results(mysql_result, cassandra_result) except Exception as e: logger.error(f"Shadow read failed: {e}") # Return MySQL result (still source of truth) return mysql_result def compare_results(self, mysql_data, cassandra_data): # Log any discrepancies if mysql_data[1] != cassandra_data.email: logger.error(f"Data mismatch for {mysql_data[1]}") # Alert ops team # ✅ Shadow reads validate data without affecting users # ✅ Monitor error rates and data discrepancies # ✅ Fix any issues before cutover

# Use feature flags to gradually shift reads to Cassandra class UserRepository: def __init__(self, mysql, cassandra, config): self.mysql = mysql self.cassandra = cassandra self.cassandra_read_percentage = config.get('cassandra_read_pct', 0) def get_user_by_email(self, email): # Randomly decide which database to read from if random.random() < (self.cassandra_read_percentage / 100): # Read from Cassandra try: result = self.cassandra.execute(""" SELECT * FROM users_by_email WHERE email = %s """, (email,)).one() return result except Exception as e: # Fallback to MySQL if Cassandra fails logger.error(f"Cassandra read failed, falling back: {e}") return self.get_from_mysql(email) else: # Read from MySQL return self.get_from_mysql(email) # Rollout schedule: # Week 1: 5% reads from Cassandra # Week 2: 25% reads from Cassandra # Week 3: 50% reads from Cassandra # Week 4: 100% reads from Cassandra # ✅ Monitor latency, error rates, throughput at each stage # ✅ Roll back percentage if issues detected

# Once 100% reads from Cassandra with no issues: # 1. Make Cassandra the source of truth for writes def create_user(self, email, username): # Now write to Cassandra FIRST user_id = uuid.uuid4() try: # Primary write (Cassandra) self.cassandra.execute("""INSERT INTO users_by_id ...""") self.cassandra.execute("""INSERT INTO users_by_email ...""") # Secondary write (MySQL, for safety) if self.mysql_write_enabled: try: self.mysql.execute("""INSERT INTO users ...""") except: logger.error("MySQL write failed (non-critical)") return user_id except Exception as e: raise e # 2. Run for 1-2 weeks monitoring everything # 3. Disable MySQL writes (keep as backup) # 4. After 30 days of stability, decommission MySQL # ✅ Migration complete!