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:

Wählen...	Wenn es am besten passt	Kompromiss
Massenkopieren mit mssql.CopyIn	Du brauchst den schnellsten Weg, viele Zeilen in einer Zieltabelle zu laden.	Bester Durchsatz, aber er zielt auf Tabellenlasten und nicht auf gespeicherte Prozedurverträge oder gemischte Nutzlasten ab.
Ein tabellenwertiger Parameter	Man muss eine stark typisierte Zeilenmenge in eine gespeicherte Prozedur oder einen parametrisierten Befehl übergeben.	Bewahrt Schema- und Prozedurgrenzen, erfordert jedoch einen benutzerdefinierten Tabellentyp und eine passende Feldreihenfolge.
JSON mit OPENJSON oder FOR JSON	Deine Anwendung tauscht bereits JSON aus, oder die Struktur der Nutzlast ist verschachtelt oder flexibel.	Portabler für App-Code, aber meist langsamer und weniger typsicher als TVPs oder Massenkopien für strukturierte Einsätze.

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:

Feld	Typ	Description
CheckConstraints	bool	Überprüfe die Einschränkungen während des Bulk-Einsatzes.
FireTriggers	bool	Lösen Sie Trigger INSERT in der Zieltabelle aus.
KeepNulls	bool	Nullwerte bewahren, anstatt Standardwerte einzufügen.
KilobytesPerBatch	int	Kilobyte pro Batch. 0 Verwendet standardmäßig den Server.
RowsPerBatch	int	Reihen pro Charge. 0 Verwendet standardmäßig den Server.
Order	[]string	ORDER Hinweis für den Ziel-Clusterindex (zum Beispiel []string{"Id ASC"}).
Tablock	bool	Fordern Sie für die Dauer des Massenkopiervorgangs eine Sperre auf Tabellenebene an.

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()
}

Tip

Wenn Sie fehlerhafte Zeilen überspringen und den Vorgang fortsetzen müssen, verwenden Sie einzelne INSERT-Anweisungen oder das Muster mit einer Staging-Tabelle: Führen Sie einen Massenimport in eine Staging-Tabelle ohne Einschränkungen durch und verwenden Sie dann ein MERGE oder INSERT...SELECT mit Fehlerbehandlung, um die Daten in die Zieltabelle zu verschieben.

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:

Methode	Relative Geschwindigkeit	Netzwerk-Roundtrips	Sperren
Individuell INSERT	Langsamster (1x)	100,000	Auf Zeilenebene pro Einfügevorgang.
Batched INSERT (1.000 Zeilen pro Aussage)	Mittel (5-10x)	100	Reihenebene pro Charge.
Massenkopieren ohne TABLOCK	Schnell (20-50x)	Das hängt von der Chargengröße ab	Massenverarbeitung auf Zeilenebene.
Massenkopieren mit TABLOCK	Schnellster (50-100x)	Das hängt von der Chargengröße ab	Sperre auf Tabellenebene, minimale Protokollierung.

Hinweis

Die tatsächliche Leistung variiert je nach Netzwerklatenz, Serverkonfiguration, Tabellenindizes und ob minimales Logging verfügbar ist. Führen Sie mit Ihrer spezifischen Workload einen Benchmark unter Verwendung von testing.B durch. Siehe Leistungsoptimierung.

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

Go-Typ	SQL Server-Spalte	Issue	Lösung
string	varchar	Implizite nvarchar Konvertierung.	Verwende den mssql.VarChar Wrapper.
float64	decimal(18,4)	Präzisionsverlust.	Als string anmelden.
time.Time	datetime2	Zeitumwandlung.	Nutze UTC-Zeiten.
nil	Jede nullierbare Spalte	Erfordert KeepNulls: true.	Legen Sie KeepNulls in BulkOptionsfest.

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