Massenoperationen mit go-mssqldb

Der go-mssqldb Treiber unterstützt Hochleistungs-Bulk-Insert-Operationen mithilfe der Funktion mssql.CopyIn. Bulk Insert umgeht den normalen Zeilen-für-Zeilen-Pfad INSERT und streamt Daten direkt zum Server mithilfe des TDS-Bulk-Copy-Protokolls.

Wähle Bulk Copy, TVP oder JSON

Verwenden Sie folgende Anleitung, wenn Sie mehrere Zeilen oder komplexe Nutzlasten an den SQL Server senden möchten:

Wenn Sie große Datenmengen in eine Staging-Tabelle oder Zieltabelle laden, beginnen Sie mit dem Massenkopieren. Wenn du gespeicherte Prozeduren mit strukturierten Reihensätzen aufrufst, fang mit TVPs an. Wenn du verschachtelte Dokumente oder lose Schemata brauchst, fang mit JSON an.

Beispiele in diesem Artikel laufen gegen die AdventureWorks2025-Beispieldatenbank . Massenkopie von Beispielen zielt auf HumanResources.Department und Production.ProductCategory.

Einfaches Bulk-Insert

Verwenden Sie mssql.CopyIn, um eine Bulk Copy-Anweisung zu erstellen, und verwenden Sie dann Exec, um Zeilen zu senden:

import (
    "database/sql"
    "log"

    "github.com/microsoft/go-mssqldb"
)

func bulkInsert(db *sql.DB) error {
    txn, err := db.Begin()
    if err != nil {
        return err
    }
    defer txn.Rollback()

    stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department", mssql.BulkOptions{},
        "Name", "GroupName"))
    if err != nil {
        return err
    }

    // Add rows
    _, err = stmt.Exec("Data Science", "Research and Development")
    if err != nil {
        return err
    }
    _, err = stmt.Exec("Cloud Ops", "Information Technology")
    if err != nil {
        return err
    }
    _, err = stmt.Exec("Developer Relations", "Sales and Marketing")
    if err != nil {
        return err
    }

    // Flush and finalize the bulk copy
    result, err := stmt.Exec()
    if err != nil {
        return err
    }

    if err = stmt.Close(); err != nil {
        return err
    }

    rowsAffected, _ := result.RowsAffected()
    log.Printf("Bulk inserted %d rows\n", rowsAffected)

    return txn.Commit()
}

Der letzte stmt.Exec() Aufruf ohne Argumente löscht alle verbleibenden Zeilen und schließt die Massenkopierung ab.

Massenoptionen

Die Struktur mssql.BulkOptions konfiguriert das Bulk-Kopierverhalten:

Beispiel mit Optionen

BulkOptions übergeben, um Constraint-Prüfungen, Trigger und Sperren zu steuern:

stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department",
    mssql.BulkOptions{
        CheckConstraints: true,
        FireTriggers:     true,
        Tablock:          true,
        RowsPerBatch:     1000,
    },
    "Name", "GroupName"))

Fehlerbehandlung

Wenn eine Zeile fehlschlägt, schlägt die gesamte Massenkopie-Operation fehl. Überprüfen Sie Fehler sowohl bei jedem Aufruf von Exec in den einzelnen Zeilen als auch beim abschließenden Flush Exec:

for _, emp := range employees {
    _, err = stmt.Exec(emp.Name, emp.GroupName)
    if err != nil {
        txn.Rollback()
        return err
    }
}

// Final flush
_, err = stmt.Exec()
if err != nil {
    txn.Rollback()
    return err
}

Fehlgeschlagene Zeilen erkennen und protokollieren

Wenn eine Massenkopie fehlschlägt, zeigt die Fehlermeldung vom SQL Server die Einschränkung oder das Datenproblem an, identifiziert aber nicht die spezifische Zeile. Um fehlgeschlagene Zeilen zu identifizieren, verwenden Sie einen Batching-Ansatz:

func bulkInsertWithRowTracking(db *sql.DB, departments []Department) error {
    txn, err := db.Begin()
    if err != nil {
        return err
    }

    stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department",
        mssql.BulkOptions{RowsPerBatch: 500}, "Name", "GroupName"))
    if err != nil {
        txn.Rollback()
        return err
    }

    for i, dept := range departments {
        _, err = stmt.Exec(dept.Name, dept.GroupName)
        if err != nil {
            txn.Rollback()
            log.Printf("Bulk copy failed at row %d (Name=%q): %v", i, dept.Name, err)
            return fmt.Errorf("bulk copy failed at row %d: %w", i, err)
        }
    }

    _, err = stmt.Exec()
    if err != nil {
        txn.Rollback()
        return fmt.Errorf("bulk copy flush failed: %w", err)
    }

    if err = stmt.Close(); err != nil {
        txn.Rollback()
        return err
    }

    return txn.Commit()
}

Stream aus CSV-Dateien

Bei großen CSV-Dateien werden Zeilen direkt aus der Datei in Massenkopien übertragen, ohne die gesamte Datei in den Speicher zu laden:

import (
    "encoding/csv"
    "io"
    "os"
)

func bulkInsertFromCSV(db *sql.DB, filePath string) error {
    f, err := os.Open(filePath)
    if err != nil {
        return err
    }
    defer f.Close()

    reader := csv.NewReader(f)

    // Skip the header row.
    _, err = reader.Read()
    if err != nil {
        return err
    }

    txn, err := db.Begin()
    if err != nil {
        return err
    }

    stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department",
        mssql.BulkOptions{Tablock: true, RowsPerBatch: 5000},
        "Name", "GroupName"))
    if err != nil {
        txn.Rollback()
        return err
    }

    var rowCount int
    for {
        record, err := reader.Read()
        if err == io.EOF {
            break
        }
        if err != nil {
            txn.Rollback()
            return fmt.Errorf("CSV read error at row %d: %w", rowCount+1, err)
        }

        _, err = stmt.Exec(record[0], record[1])
        if err != nil {
            txn.Rollback()
            return fmt.Errorf("row %d: %w", rowCount+1, err)
        }
        rowCount++
    }

    // Flush remaining rows.
    result, err := stmt.Exec()
    if err != nil {
        txn.Rollback()
        return err
    }
    if err = stmt.Close(); err != nil {
        txn.Rollback()
        return err
    }

    affected, _ := result.RowsAffected()
    log.Printf("Bulk inserted %d rows from CSV", affected)

    return txn.Commit()
}

Leistungsvergleich

Massenkopien sind bei großen Datenmengen deutlich schneller als einzelne Einsätze. Die folgende Tabelle zeigt ungefähre Leistungsmerkmale für das Einfügen von 100.000 Zeilen:

Spaltenreihenfolge und Typzuordnung

Spalten in mssql.CopyIn müssen der Reihenfolge und den Typen entsprechen, die von der Zieltabelle erwartet werden. Der Treiber führt keinen Abgleich von Spaltennamen durch; er verwendet eine positionsbasierte Zuordnung.

Häufige Typprobleme

Beispiel mit expliziten Typen

Spezifiziere explizit Spaltentypen, wenn die Standardtypzuordnung nicht mit deinem Schema übereinstimmt:

stmt, err := txn.Prepare(mssql.CopyIn("Production.ProductCategory",
    mssql.BulkOptions{KeepNulls: true},
    "Name"))
if err != nil {
    return err
}

for _, p := range categories {
    _, err = stmt.Exec(p.Name)
    if err != nil {
        return err
    }
}

Leistungstipps

• Verwenden Sie Tablock für große Einfügungen bei leeren Tabellen. Diese Option reduziert Sperrkonflikte und ermöglicht eine minimale Protokollierung.
• Set RowsPerBatch um zu steuern, wie oft der Treiber Daten sendet. Größere Stapel verringern die Anzahl der Hin- und Rückwege, benötigen aber mehr Speicher.
• Erhöhen Sie packet size in der Verbindungszeichenfolge (bis auf 32767), um den Netzwerk-Overhead zu reduzieren.
• Ordne die Daten so, dass sie mit dem Cluster-Index der Zieltabelle übereinstimmen, und setze die Order Option. Dieser Ansatz vermeidet eine serverseitige Sortierung.
• Nicht gruppierte Indizes löschen Sie vor großen Massenladevorgängen und erstellen Sie sie anschließend neu. Die Indexwartung während des Masseneinfügens verursacht zusätzlichen Aufwand.
• Verwenden Sie CheckConstraints: false (die Standardeinstellung) für vertrauenswürdige Daten, um die Einschränkungsprüfung während der Massenkopie zu überspringen.

Einschränkungen

• Massenkopien unterstützen keine Spalten, die durch Always Encrypted geschützt sind. Weitere Informationen finden Sie unter "Einschränkungen".
• Bulk-Copy auf TDS-Ebene wird in Azure SQL-Datenbank nicht unterstützt. Für Azure SQL-Datenbank verwenden Sie stattdessen Batch-Anweisungen INSERT oder ein Staging Table Pattern.

Verwandte Inhalte

• Leistungsoptimierung
• Abfragen und Aussagen
• Datentypzuordnungen
• Migrationshandbuch
• Fehlerbehandlung und Wiederholungsmuster