単一の MySQL テーブルで数千万のデータを処理するアイデアを共有する

db.tb から a、b、c を選択 制限 10000 オフセット 9000000

id=9999999 の db.tb から a、b、c を選択します。
db.tb を更新し、a=x、id=9999999 に設定します。

定義ミッションハンドラー(すべてのミッション、ワーカーミッションサイズ):
    「」
    タスク リストは、タスクの合計数と各ワーカーのタスク数に基づいて計算されます。タスク リストの要素は (タスク開始 ID、タスク終了 ID) です。
    例: タスクの総数は 100 で、各ワーカーのタスク数は 40 です。この場合、タスク リストは次のようになります: [(1, 40), (41, 80), (81, 100)]
    :param all_missions: ミッションの総数 :param worker_mission_size: 各ワーカーの最大ミッション数 :return: [(start_id, end_id), (start_id, end_id), ...]
    「」
    ワーカーミッションID = []
    現在のID = 0
    current_id <= all_missions の場合:
        start_id = all_missions if current_id + 1 >= all_missions それ以外の場合は current_id + 1
        end_id = all_missions、if current_id + worker_mission_size >= all_missions、そうでない場合は current_id + worker_mission_size
        start_id == end_idの場合:
            ワーカーミッションID[-1][1] == 開始IDの場合:
                壊す
        ワーカーミッションIDを追加します((開始ID、終了ID))
        現在のID += ワーカーミッションサイズ

    ワーカーミッションIDを返す

>>> ミッションハンドラ(100, 40)
[(1, 40), (41, 80), (81, 100)]

並行.futures から ProcessPoolExecutor をインポートします


main() を定義します:
    # 自動増分 ID の最大値 max_id = 30000000
    # 単一のワーカーによって処理されるデータ量 worker_mission_size = 1000000
    # 複数のプロセスを使用してミッションを処理する = mission_handler(max_id, worker_mission_size)
    労働者 = []
    実行者 = ProcessPoolExecutor()
    idxの場合、enumerate(missions)のミッション:
        start_id、end_id = ミッション
        ワーカー.append(executor.submit(データハンドラー、開始ID、終了ID、idx))


データハンドラを定義します(開始ID、終了ID、ワーカーID):
    合格

# 処理ステータスを記録するために別のテーブルを使用します insert into db.tb_handle_status(row_id, success) values ​​(999, 0);

データハンドラを定義します(開始ID、終了ID、ワーカーID):
    #データ接続 conn、カーソル = mysql()
    現在のID = 開始ID
        試す：
            current_id <= end_id の場合:
                試す：
                    # TODO データ処理コードパス

                except 例外を e として:
                    # TODOレコード処理結果# データは次のcurrent_id += 1に移動する
                    続く
                それ以外：
                    # 例外なし、次のデータの処理を続行 current_id += 1
        except 例外を e として:
            'worker_id({}): result({})'.format(worker_id, False) を返します。
        ついに：
            # データベースリソースの解放 cursor.close()
            接続を閉じる()

        'worker_id({}): result({})' を返します。format(worker_id, True)

sql = """db.tb を更新し、a=%s、b=%s を設定します (ID=%s の場合)"""
値 = [
            ('a_value', 'b_value', 9999)、
            ('a_value'、'b_value'、9998)、
            ...
         ]
# ネットワーク IO とロック取得頻度を削減するためのバッチ送信 cursor.executemany(sql, values)