1. query.log

Um MySQL die Protokollierung aller ausgeführten Abfragen (das sogenannte allgemeine Abfrageprotokoll) zu ermöglichen, müssen Sie die entsprechende globale Variable aktivieren. Dieses Protokoll ist standardmäßig deaktiviert, da es auf laufenden Datenbanken schnell viel Speicherplatz belegt. Sie können es auf zwei Arten aktivieren: temporär (bis zum Neustart des Servers) oder dauerhaft (über die Konfigurationsdatei).

sudo -s
mysql
              

Methode 1: Temporäres Aktivieren (ohne MySQL neu zu starten) Diese Option ist ideal, wenn Sie nur 10-15 Minuten lang Protokolle für die Analyse durch ein Skript sammeln und sie dann wieder deaktivieren müssen.

SET GLOBAL general_log_file = '/var/log/mysql/query.log';
              

Wichtig: Stellen Sie sicher, dass der MySQL-Betriebssystembenutzer Schreibzugriff auf das angegebene Verzeichnis hat (/var/log/mysql/). Im Zweifelsfall können Sie den Standardpfad beibehalten, indem Sie einfach den zweiten Befehl ausführen – die Datei wird dann im Datenverzeichnis (datadir) erstellt.

Um zu überprüfen, ob alles eingeschaltet ist, führen Sie Folgendes aus:

SHOW VARIABLES LIKE '%general_log%';
              

Wie man es wieder ausschaltet:

SET GLOBAL general_log = 'OFF';
              

Methode 2: Dauerhaft aktivieren (über Konfigurationsdatei)

Falls Sie ständig Protokolle benötigen oder die obige Methode nicht geeignet ist, müssen die Einstellungen in die Konfigurationsdatei geschrieben werden. Öffnen Sie die Konfigurationsdatei MySQL (normalerweise /etc/mysql/my.cnf, /etc/my.cnf oder /etc/mysql/mysql.conf.d/mysqld.cnf).

[mysqld]
general_log      = 1
general_log_file = /var/log/mysql/query.log
              

Speichern Sie die Datei und starten Sie den MySQL-Dienst neu, damit die Einstellungen wirksam werden.:

systemctl restart mysql
# oder für ältere Systeme / MariaDB:
systemctl restart mariadb
              

Logrotate-Konfigurationsdatei erstellen

tee /etc/logrotate.d/mysql > /dev/null << 'EOF'
/var/log/mysql/query.log /var/log/mysql/mysql.log /var/log/mysql/slow.log {
    daily
    rotate 7
    missingok
    notifempty
    compress
    delaycompress
    sharedscripts
    create 660 mysql mysql
    postrotate
        if test -x /usr/bin/mysqladmin && /usr/bin/mysqladmin ping &>/dev/null; then
            /usr/bin/mysqladmin --local flush-general-log flush-slow-log
        fi
    endscript
}
EOF
              

Test

logrotate -d /etc/logrotate.d/mysql
              

2. virtual environment (venv)

Am besten arbeitet man in einer isolierten Umgebung. Ich persönlich bevorzuge venv, aber man kann natürlich auch conda, miniconda usw. verwenden.

python3 -m venv .venv
source .venv/bin/activate
              

3. requirements.txt

Alle notwendigen Bibliotheken befinden sich hier.

sqlparse>=0.5.0
              

4. Installation

Ich empfehle, genau die in der Datei recommendations.txt angegebenen Versionen zu installieren.

pip install -r requirements.txt
              

5. Migration-Script

LogParser verwendet yield, anstatt die gesamte Datei in den Arbeitsspeicher zu laden. Dies ist entscheidend, da die entpackte Datei general_log mehrere Gigabyte groß sein kann. Das Skript speichert nur eine Zeile im Speicher und erhöht Zähler. Sie können entweder query.log oder die archivierte Datei query.log.2.gz (die von logrotate gepackt wird) an das Argument -i übergeben. Das Skript erkennt das Format automatisch anhand der Dateiendung.

import os
import sys
import gzip
import argparse
import sqlparse
from sqlparse.sql import IdentifierList, Identifier, Comparison
from collections import Counter, defaultdict

class LogParser:
    """Die Klasse ist ausschließlich für das Öffnen, Lesen und Streamen von Zeichenketten aus Dateien (einschließlich .gz-Dateien) zuständig."""
    def __init__(self, file_path):
        self.file_path = file_path

    def stream_lines(self):
        """Öffnet eine Datei (normal oder gzip) und gibt die Zeilen nacheinander zurück."""
        if not os.path.exists(self.file_path):
            print(f"[-] Fehler: Datei {self.file_path} nicht gefunden.", file=sys.stderr)
            return

        # Überprüfung der Dateierweiterung auf gzip-Unterstützung
        is_gzip = self.file_path.endswith('.gz')
        open_func = gzip.open if is_gzip else open
        mode = 'rt' if is_gzip else 'r' # rt - Text für gzip lesen

        print(f"[+] Öffnen einer Datei: {self.file_path}")

        try:
            with open_func(self.file_path, mode, encoding='utf-8', errors='ignore') as f:
                for line in f:
                    yield line
        except Exception as e:
            print(f"[-] Fehler beim Lesen der Datei: {e}", file=sys.stderr)


class SchemaAnalyzer:
    """Die Klasse ist für das Parsen von SQL-Abfragen und das Sammeln von Statistiken zuständig."""
    def __init__(self):
        # (TableA, TableB) -> count
        self.join_frequency = Counter()
        # Table -> {Column: count}
        self.where_columns = defaultdict(Counter)

    def analyze_line(self, line):
        """Löscht die Logzeile und sendet die SQL-Anweisung zur Analyse."""
        clean_line = line.strip()
        # MySQL-Logkommentare und leere Zeilen werden übersprungen
        if not clean_line or clean_line.startswith(('#', '--', '/*')):
            return False

        try:
            parsed = sqlparse.parse(clean_line)
            for stmt in parsed:
                if stmt.get_type() != 'SELECT':
                    continue

                # 1. JOIN-Verbindungen fixen
                tables = self._extract_tables(stmt)
                if len(tables) > 1:
                    pair = tuple(sorted(tables))
                    self.join_frequency[pair] += 1

                # 2. Fixen WHERE-Bedingungen
                self._extract_where_conditions(stmt)
            return True
        except Exception:
            # Ignoriere syntaktisch fehlerhafte Logzeilen.
            return False

    def _clean_name(self, name):
        if not name:
            return ""
        return name.replace('`', '').replace('"', '').replace("'", "").strip().lower()

    def _extract_tables(self, stmt):
        tables = []
        for token in stmt.tokens:
            if isinstance(token, Identifier):
                real_name = token.get_real_name()
                if real_name:
                    tables.append(self._clean_name(real_name))
            elif isinstance(token, IdentifierList):
                for identifier in token.get_identifiers():
                    real_name = identifier.get_real_name()
                    if real_name:
                        tables.append(self._clean_name(real_name))
        return list(set(tables))

    def _extract_where_conditions(self, stmt):
        where_token = None
        for token in stmt.tokens:
            if token.is_keyword and token.value.upper() == 'WHERE':
                where_token = token
                break

        if not where_token:
            return

        idx = stmt.token_index(where_token)
        for token in stmt.tokens[idx:]:
            if isinstance(token, Comparison):
                col_raw = token.left.value
                table_name = "global"

                if '.' in col_raw:
                    parts = col_raw.split('.')
                    table_name = self._clean_name(parts[0])
                    col_name = self._clean_name(parts[1])
                else:
                    col_name = self._clean_name(col_raw)

                if col_name:
                    self.where_columns[table_name][col_name] += 1


class CassandraSchemaGenerator:
    """Die Klasse ist ausschließlich dafür zuständig, die gesammelten Statistiken in das Cassandra CQL-Schema zu übersetzen."""
    def __init__(self, analyzer: SchemaAnalyzer, min_join_threshold=5):
        self.analyzer = analyzer
        self.min_join_threshold = min_join_threshold

    def generate_cql(self, keyspace_name="generated_keyspace"):
        cql_statements = []
        cql_statements.append(
            f"CREATE KEYSPACE IF NOT EXISTS {keyspace_name} \n"
            f"WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': 1}};"
        )
        cql_statements.append(f"USE {keyspace_name};\n")

        denormalized_tables = set()

        # 1. Denormalisierung häufiger Joins
        for pair, count in self.analyzer.join_frequency.items():
            if count >= self.min_join_threshold:
                table_a, table_b = pair
                denormalized_name = f"{table_a}_by_{table_b}"
                denormalized_tables.add(table_a)
                denormalized_tables.add(table_b)

                clustering_cols = list(self.analyzer.where_columns[table_a].keys()) + list(self.analyzer.where_columns[table_b].keys())
                clustering_cols = list(dict.fromkeys(clustering_cols))

                cql = self._build_table_cql(denormalized_name, clustering_cols, is_denormalized=True, base_tables=pair)
                cql_statements.append(cql)

        # 2. Isolierte Tabellen (wenige oder keine JOINs)
        all_tables = set(self.analyzer.where_columns.keys()) - {'global', 'unknown'}
        standalone_tables = all_tables - denormalized_tables

        for table in standalone_tables:
            clustering_cols = list(self.analyzer.where_columns[table].keys())
            cql = self._build_table_cql(table, clustering_cols)
            cql_statements.append(cql)

        return "\n\n".join(cql_statements)

    def _build_table_cql(self, table_name, clustering_cols, is_denormalized=False, base_tables=None):
        partition_key = f"{base_tables[1]}_id" if is_denormalized else "id"
        clustering_cols = [c for c in clustering_cols if c != partition_key and c != 'id']

        columns = {partition_key: "uuid"}

        for col in clustering_cols:
            if any(x in col for x in ['date', 'time', 'ts', 'created', 'updated']):
                columns[col] = "timestamp"
            elif 'id' in col:
                columns[col] = "uuid"
            elif any(x in col for x in ['count', 'status', 'num', 'qty', 'is_']):
                columns[col] = "int"
            elif any(x in col for x in ['price', 'amount', 'sum', 'balance']):
                columns[col] = "decimal"
            else:
                columns[col] = "text"

        if is_denormalized:
            columns[f"{base_tables[0]}_payload"] = "text"
            columns[f"{base_tables[1]}_payload"] = "text"
        else:
            columns["data_payload"] = "text"

        col_definitions = [f"    {col} {dtype}" for col, dtype in columns.items()]

        if clustering_cols:
            pk_string = f"PRIMARY KEY (({partition_key}), {', '.join(clustering_cols)})"
        else:
            pk_string = f"PRIMARY KEY ({partition_key})"

        col_definitions.append(f"    {pk_string}")

        cql = f"CREATE TABLE IF NOT EXISTS {table_name} (\n" + ",\n".join(col_definitions) + "\n)"

        if clustering_cols:
            order_by = ", ".join([f"{col} DESC" for col in clustering_cols])
            cql += f" WITH CLUSTERING ORDER BY ({order_by});"
        else:
            cql += ";"

        return cql


def main():
    parser = argparse.ArgumentParser(description="MySQL-Log-Analyzer (einschließlich .gz) zur Generierung der Cassandra-DB-Architektur.")
    parser.add_argument("-i", "--input", required=True, help="Pfad zur Protokolldatei (z. B. query.log oder query.log.1.gz).")
    parser.add_argument("-o", "--output", default="schema.cql", help="Pfad zum Speichern der fertigen CQL-Schemadatei.")
    parser.add_argument("-t", "--threshold", type=int, default=5, help="JOIN-Übereinstimmungsschwelle für die Denormalisierung.")
    parser.add_argument("-k", "--keyspace", default="generated_keyspace", help="Der Name des in Cassandra erstellten Keyspace.")

    args = parser.parse_args()

    # 1. Initialisiere den Parser für eine bestimmte Datei
    log_parser = LogParser(args.input)

    # 2. Initialisierung eines leeren Strukturanalysators
    analyzer = SchemaAnalyzer()

    # Wir beginnen, Datenströme vom Parser zum Analysator zu übertragen
    processed_lines = 0
    success_queries = 0

    for line in log_parser.stream_lines():
        processed_lines += 1
        if analyzer.analyze_line(line):
            success_queries += 1

    print(f"[+] Protokollanalyse abgeschlossen. Gelesene Zeilen: {processed_lines}, Gültige SELECT-Abfragen: {success_queries}")

    if success_queries == 0:
        print("[-] Warnung: Im Protokoll wurden keine geeigneten SELECT-Abfragen zum Erstellen des Schemas gefunden.")
        sys.exit(0)

    # 3. Wir leiten den Analysator an den Schaltungsgenerator weiter.
    generator = CassandraSchemaGenerator(analyzer, min_join_threshold=args.threshold)
    cql_schema = generator.generate_cql(keyspace_name=args.keyspace)

    # Ergebnis speichern
    try:
        with open(args.output, 'w', encoding='utf-8') as f:
            f.write(cql_schema)
        print(f"[+] Cassandra-Schema erfolgreich generiert und gespeichert unter: {args.output}")
    except Exception as e:
        print(f"[-] Fehler beim Schreiben der Ausgabedatei: {e}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()