This implementation focuses on migrating MySQL data to a MongoDB sharded cluster using Python, with adaptations for older MongoDB 3.4 APIs and enhancements for multithreading and streaming. The original code from an open-source repository has been modified to support concurrent processing and compatibility with PyMongo's legacy methods.

import sys
import mysql.connector
import pymongo
import datetime
import enum
import numpy
import threading

class LogColors:
    HEADER = '\033[95m'
    INFO = '\033[94m'
    DEBUG = '\033[96m'
    SUCCESS = '\033[92m'
    WARNING = '\033[93m'
    ERROR = '\033[91m'
    RESET = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

class LogLevel(enum.Enum):
    HEADER = 1
    INFO = 2
    DEBUG = 3
    SUCCESS = 4
    WARNING = 5
    ERROR = 6
    RESET = 7
    BOLD = 8
    UNDERLINE = 9

def log_message(message, level):
    if level == LogLevel.HEADER:
        print(f"{LogColors.HEADER}{message}{LogColors.RESET}")
    elif level == LogLevel.INFO:
        print(f"{LogColors.INFO}{message}{LogColors.RESET}")
    elif level == LogLevel.DEBUG:
        print(f"{LogColors.DEBUG}{message}{LogColors.RESET}")
    elif level == LogLevel.SUCCESS:
        print(f"{LogColors.SUCCESS}{message}{LogColors.RESET}")
    elif level == LogLevel.WARNING:
        print(f"{LogColors.WARNING}{message}{LogColors.RESET}")
    elif level == LogLevel.ERROR:
        print(f"{LogColors.ERROR}{message}{LogColors.RESET}")
    elif level == LogLevel.BOLD:
        print(f"{LogColors.BOLD}{message}{LogColors.RESET}")
    elif level == LogLevel.UNDERLINE:
        print(f"{LogColors.UNDERLINE}{message}{LogColors.RESET}")

def transfer_table(mysql_host, mysql_db, mysql_user, mysql_pass, table_name, mongo_collection, clear_existing):
    mysql_conn = mysql.connector.connect(
        host=mysql_host,
        database=mysql_db,
        user=mysql_user,
        password=mysql_pass
    )
    cursor = mysql_conn.cursor(dictionary=True)
    cursor.execute(f"SELECT * FROM {table_name}")
    records = cursor.fetchall()
    
    if clear_existing:
        mongo_collection.delete_many({})
    
    if records:
        result = mongo_collection.insert_many(records)
        return len(result.inserted_ids)
    return 0

start_time = datetime.datetime.now()
canceled = False
log_message(f"Migration initiated at: {start_time}", LogLevel.HEADER)

clear_docs = True
mysql_host = "192.168.1.100"
mysql_database = "source_db"
mysql_schema = "public"
mysql_user = "admin"
mysql_password = "securepass"

mongo_uri = "mongodb://user:pass@192.168.1.101:27017/admin"
mongo_db_name = "target_db"

if clear_docs:
    response = input("Remove existing MongoDB documents? (y)es/(n)o/(c)ancel: ")
    if response.lower() == "c":
        canceled = True
    elif response.lower() == "n":
        clear_docs = False
    else:
        confirm = input("Confirm deletion (y)es/(n)o: ")
        if confirm.lower() == "y":
            clear_docs = True
        else:
            canceled = True

if canceled:
    log_message("Migration canceled by user", LogLevel.ERROR)
else:
    if clear_docs:
        log_message("Existing documents will be removed", LogLevel.ERROR)
    else:
        log_message("Existing documents preserved", LogLevel.SUCCESS)

    log_message("Establishing MySQL connection...", LogLevel.HEADER)
    mysql_db = mysql.connector.connect(
        host=mysql_host,
        database=mysql_database,
        user=mysql_user,
        password=mysql_password
    )
    log_message("MySQL connection successful", LogLevel.SUCCESS)

    log_message("Establishing MongoDB connection...", LogLevel.HEADER)
    mongo_client = pymongo.MongoClient(mongo_uri)
    mongo_db = mongo_client[mongo_db_name]
    log_message("MongoDB connection successful", LogLevel.SUCCESS)

    log_message("Starting data transfer...", LogLevel.HEADER)
    db_list = mongo_client.database_names()
    if mongo_db_name in db_list:
        log_message("Target database exists", LogLevel.INFO)
    else:
        log_message("Creating new database", LogLevel.WARNING)

    schema_cursor = mysql_db.cursor()
    schema_cursor.execute(
        "SELECT table_name FROM information_schema.tables WHERE table_schema = %s ORDER BY table_name LIMIT 15;",
        (mysql_schema,)
    )
    tables = schema_cursor.fetchall()

    total_tables = len(tables)
    log_message(f"Tables to migrate: {total_tables}", LogLevel.SUCCESS)
    completed = 0
    failed = 0

    reshaped_tables = numpy.array(tables).reshape(2, int(len(tables) / 2))
    for table_pair in reshaped_tables:
        try:
            log_message(f"Processing table: {table_pair[0]}...", LogLevel.DEBUG)
            thread_a = threading.Thread(
                target=transfer_table,
                args=(mysql_host, mysql_database, mysql_user, mysql_password, table_pair[0], mongo_db[table_pair[0]], clear_docs)
            )
            thread_a.start()
            
            log_message(f"Processing table: {table_pair[1]}...", LogLevel.DEBUG)
            thread_b = threading.Thread(
                target=transfer_table,
                args=(mysql_host, mysql_database, mysql_user, mysql_password, table_pair[1], mongo_db[table_pair[1]], clear_docs)
            )
            thread_b.start()
            
            thread_a.join()
            thread_b.join()
            completed += 2
            finish_time = datetime.datetime.now()
            log_message(f"Table {table_pair[0]} completed at: {finish_time}", LogLevel.SUCCESS)
            log_message(f"Table {table_pair[1]} completed at: {finish_time}", LogLevel.SUCCESS)
        except Exception as err:
            failed += 2
            log_message(f"Error: {err}", LogLevel.ERROR)

    log_message("Migration finished", LogLevel.HEADER)
    log_message(f"Successfully migrated {completed} of {total_tables} tables", LogLevel.SUCCESS)
    if failed > 0:
        log_message(f"Failed to migrate {failed} tables. Review errors above.", LogLevel.ERROR)

end_time = datetime.datetime.now()
log_message(f"Script completed at: {end_time}", LogLevel.HEADER)
log_message(f"Total duration: {end_time - start_time}", LogLevel.HEADER)

Performance analysis shows that migrating 6 million rows (400 MB) from MySQL to a two-shard MongoDB cluster takes approximately 45 minutes in single-threaded mode, largely due to shard balancing overhead. Without sharding, the same operation cmopletes in about 18 minutes.

Multithreaded execution with two threads processing four tables (600,000 rows each) requires 1 hour 52 minutes, averaging 30 minutes per table. A streaming version using server-side cursors was tested but proved less efficient, taking 4 hours for four tables with four concurrent threads, compared to 2 hours for the non-streaming approach with two threads.

Streaming implementations using pymysql.cursors.SSDictCursor can encounter connection timeouts and resource management issues, such as:

• Lost MySQL connections during extended queries
• Incomplete cursor closures leading to warnings

These can be mitigated with proper error handling and connection pooling, but batch insertion generally outperforms row-by-row streaming for bulk migrations.

Python offers multiple MySQL connectors:

• mysql-connector-python: Official pure-Python driver with good portability
• MySQLdb: C-based wrapper offering higher performance but less portability
• pymysql: Pure-Python alternative with similar characteristics to the official driver

For large datasets, server-side cursors with SSCursor or SSDictCursor prevent memory overload by streaming results incrementally. However, they maintian active connections that block other operations and require timely processing to avoid network timeouts.