Austausch von Ideen zur Verarbeitung von Millionen von Daten in einer einzigen MySQL-Tabelle

wähle a,b,c aus db.tb Limit 10000 Offset 9000000

Wählen Sie a, b, c aus db.tb, wobei id=9999999;
Aktualisiere db.tb, setze a=x, wobei id=9999999;

def Missionshandler (alle Missionen, Arbeiter-Missionsgröße):
    """
    Die Aufgabenliste wird basierend auf der Gesamtzahl der Aufgaben und der Anzahl der Aufgaben jedes Mitarbeiters berechnet. Die Elemente der Aufgabenliste sind (Aufgabenstart-ID, Aufgabenende-ID).
    Beispiel: Die Gesamtzahl der Aufgaben beträgt 100 und die Anzahl der Aufgaben für jeden Mitarbeiter beträgt 40. Dann lautet die Aufgabenliste: [(1, 40), (41, 80), (81, 100)]
    :param all_missions: Gesamtzahl der Missionen :param worker_mission_size: maximale Anzahl der Missionen für jeden Arbeiter :return: [(start_id, end_id), (start_id, end_id), ...]
    """
    Arbeiter_Missions-IDs = []
    aktuelle_id = 0
    während aktuelle_id <= alle_missionen:
        start_id = all_missions wenn current_id + 1 >= all_missions sonst current_id + 1
        end_id = alle_Missionen wenn aktuelle_ID + Arbeiter-Missionsgröße >= alle_Missionen sonst aktuelle_ID + Arbeiter-Missionsgröße
        wenn start_id == end_id:
            wenn worker_mission_ids[-1][1] == start_id:
                brechen
        worker_mission_ids.anhängen((start_id, end_id))
        aktuelle_ID += Arbeiter-Missionsgröße

    Arbeiter-Missions-IDs zurückgeben

>>> Missionshandler (100, 40)
[(1, 40), (41, 80), (81, 100)]

von concurrent.futures importiere ProcessPoolExecutor


def main():
    # Maximalwert der Auto-Increment-ID max_id = 30000000
    # Von einem einzelnen Worker verarbeitetes Datenvolumen worker_mission_size = 1000000
    # Verwenden Sie mehrere Prozesse, um Missionen zu verarbeiten = Missionshandler (max_id, Worker_Missionsgröße).
    Arbeiter = []
    Executor = ProcessPoolExecutor()
    für idx, Mission in enumerate(Missionen):
        Start-ID, End-ID = Mission
        Arbeiter.Anhängen(Executor.Senden(Datenhandler, Start-ID, End-ID, IDX))


auf def data_handler(start_id, end_id, worker_id):
    passieren

# Verwenden Sie eine andere Tabelle, um den Verarbeitungsstatus aufzuzeichnen. Einfügen in db.tb_handle_status (row_id, success) Werte (999, 0);

auf def data_handler(start_id, end_id, worker_id):
    #Datenverbindung conn, cursor = mysql()
    aktuelle_id = start_id
        versuchen:
            während aktuelle_id <= end_id:
                versuchen:
                    # TODO Datenverarbeitungscode-Pass

                außer Ausnahme als e:
                    # TODO-Datensatzverarbeitungsergebnisse# Daten werden zur nächsten current_id += 1 verschoben.
                    weitermachen
                anders:
                    # Keine Ausnahme, fahren Sie mit der Verarbeitung der nächsten Daten fort current_id += 1
        außer Ausnahme als e:
            gibt 'worker_id({})' zurück: result({})'.format(worker_id, False)
        Endlich:
            # Datenbankressourcenfreigabe cursor.close()
            conn.close()

        gibt 'worker_id({})' zurück: result({})'.format(worker_id, True)

sql = """Aktualisiere db.tb, setze a=%s, b=%s, wobei id=%s"""
Werte = [
            ('a_Wert', 'b_Wert', 9999),
            ('a_Wert', 'b_Wert', 9998),
            ...
         ]
# Batch-Übermittlung zur Reduzierung der Netzwerk-E/A und der Häufigkeit des Sperrenerwerbs cursor.executemany(sql, values)