Operacje masowe z go-mssqldb

Sterownik go-mssqldb obsługuje wysokowydajne operacje wstawiania zbiorczego za pomocą funkcji mssql.CopyIn. Bulk insert omija normalną ścieżkę wiersz po wierszu INSERT i przesyła dane bezpośrednio do serwera, korzystając z protokołu kopiowania masowego TDS.

Wybierz kopię masową, TVP lub JSON

Skorzystaj z poniższego przewodnika, gdy musisz wysłać wiele wierszy lub złożone ładunki do SQL Server:

Wybierz...	Kiedy najlepiej pasuje	Kompromis
Kopiowanie zbiorcze z mssql.CopyIn	Potrzebujesz najszybszego sposobu na załadowanie wielu wierszy do jednej tabeli docelowej.	Najlepsza przepustowość, ale jest ukierunkowana na ładowanie tabel, a nie na kontrakty procedur składowanych ani ładunki o mieszanej strukturze.
Parametr tabelowy	Musisz przekazać silnie wpisany zestaw wierszy do procedury zapisanej lub polecenia parametryzowanego.	Zachowuje granice schematu i procedur, ale wymaga użytkownika zdefiniowanego typu tabeli i odpowiadającego kolejności pól.
JSON z OPENJSON lub FOR JSON	Twoja aplikacja już wymienia dane w formacie JSON albo struktura ładunku danych jest zagnieżdżona lub zmienna.	Bardziej przenośny dla kodu aplikacji, ale zwykle wolniejszy i mniej bezpieczny pod względem pisania niż TVP lub kopia masowa do wstawek strukturalnych.

Jeśli ładujesz duże partie danych do tabeli tymczasowej lub docelowej, zacznij od operacji kopiowania zbiorczego. Jeśli wywołujesz procedury przechowywane z ustrukturyzowanymi zestawami wierszy, zacznij od TVP-ów. Jeśli potrzebujesz zagnieżdżonych dokumentów lub luźnych schematów, zacznij od JSON.

Przykłady w tym artykule porównane są z przykładową bazą danych AdventureWorks2025 . Przykłady zbiorczego kopiowania dotyczą HumanResources.Department i Production.ProductCategory.

Podstawowa wkładka zbiorcza

Użyj mssql.CopyIn, aby utworzyć instrukcję zbiorczego kopiowania, a następnie użyj Exec, aby wysłać wiersze:

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

Ostatnie stmt.Exec() wywołanie bez argumentów usuwa pozostałe wiersze i kończy operację kopiowania masowego.

Opcje zbiorcze

Struktura mssql.BulkOptions konfiguruje zachowanie kopiowania masowego:

Pole	Typ	Description
CheckConstraints	bool	Sprawdź ograniczenia podczas wstawiania zbiorczego.
FireTriggers	bool	Spusty ognia INSERT na stole celu.
KeepNulls	bool	Zachowaj wartości null zamiast wstawiać wartości domyślne.
KilobytesPerBatch	int	Kilobajty na partię. 0 używa domyślnego serwera.
RowsPerBatch	int	Liczba wierszy na wsad. 0 używa domyślnego serwera.
Order	[]string	ORDER wskazówka dla docelowego indeksu klastrowanego (na przykład []string{"Id ASC"}).
Tablock	bool	Uzyskaj blokadę na poziomie tabeli na czas trwania kopiowania zbiorczego.

Przykład z opcjami

Przekaż BulkOptions, aby kontrolować sprawdzanie ograniczeń, wyzwalacze i blokowanie:

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

Obsługa błędów

Jeśli w dowolnym wierszu wystąpi błąd, cała operacja kopiowania zbiorczego zakończy się niepowodzeniem. Sprawdź błędy zarówno z wywołania Exec każdego rzędu, jak i z końcowego flushu 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
}

Wykryj i zarejestruj wiersze zakończone niepowodzeniem

Gdy operacja kopiowania masowego nie udaje się, komunikat o błędzie z SQL Server wskazuje na ograniczenie lub problem z danymi, ale nie identyfikuje konkretnego wiersza. Aby zidentyfikować nieudane wiersze, zastosuj podejście grupowania:

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

Wskazówka

Jeśli musisz pominąć błędne wiersze i kontynuować, użyj pojedynczych instrukcji INSERT lub wzorca tabeli przejściowej: wykonaj kopiowanie zbiorcze do tabeli przejściowej bez ograniczeń, a następnie użyj instrukcji MERGE lub INSERT...SELECT z obsługą błędów, aby przenieść dane do tabeli docelowej.

Stream z plików CSV

W przypadku dużych plików CSV przesyłaj wiersze bezpośrednio z pliku do kopii masowej bez ładowania całego pliku do pamięci:

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

Porównanie wydajności

Kopiowanie masowe jest znacznie szybsze niż pojedyncze wstawiania przy dużych ładunkach danych. Poniższa tabela przedstawia przybliżone charakterystyki wydajności przy wstawianiu 100 000 wierszy:

Metoda	Prędkość względna	Obiegi sieciowe	Blokowanie
Indywidualny INSERT	Najwolniejszy (1x)	100,000	Poziom wiersza na każdą wstawkę.
Batched INSERT (1000 wierszy na każde zdanie)	Średnia (5-10x)	100	Poziom rzędów na partię.
Kopia masowa bez TABLOCK	Szybki (20-50x)	To zależy od wielkości partii	Bulk na poziomie rzędu.
Kopiowanie zbiorcze z TABLOCK	Najszybszy (50-100x)	To zależy od wielkości partii	Blokada na poziomie stołu, minimalne rejestrowanie.

Note

Rzeczywista wydajność zależy od opóźnień sieci, konfiguracji serwera, indeksów tabel oraz tego, czy dostępne jest minimalne logowanie. Porównaj z twoim konkretnym obciążeniem za pomocą testing.B. Zobacz Dostrajanie wydajności.

Kolejność kolumn i mapowanie typów

Kolumny w mssql.CopyIn muszą odpowiadać kolejności i typom oczekiwanym przez docelową tabelę. Sterownik nie wykonuje dopasowywania nazw kolumn; wykorzystuje mapowanie pozycyjne.

Typowe problemy

typ Go	Kolumna SQL Server	Issue	Rozwiązanie
string	varchar	Ukryta nvarchar konwersja.	Użyj elementu mssql.VarChar wrapper.
float64	decimal(18,4)	Strata precyzji.	Passuj jako string.
time.Time	datetime2	Konwersja stref czasowych.	Używaj czasów UTC.
nil	Dowolna kolumna nulowalna	Wymaga KeepNulls: true.	Ustaw KeepNulls w pliku BulkOptions.

Przykład z typami jawnymi

Określ typy kolumn jawnie, gdy domyślne mapowanie typów nie odpowiada twojemu schematowi:

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

Porady dotyczące wydajności

• Zastosowanie Tablock dla dużych wstawek do pustych tabel. Ta opcja zmniejsza rywalizację o blokady i umożliwia minimalne logowanie.
• Set RowsPerBatch Aby kontrolować, jak często kierowca przesyła dane. Większe partie zmniejszają liczbę podróży w obie strony, ale zużywają więcej pamięci.
• Zwiększ packet size w parametrach połączenia (maksymalnie do 32767), aby zmniejszyć narzut sieciowy.
• Posortuj dane tak, aby odpowiadały klastrowanemu indeksowi tabeli docelowej, i ustaw opcję Order. Takie podejście unika sortowania po stronie serwera.
• Zrzucaj indeksy nieklastrowane przed dużymi ładunkami masowymi, a potem je odbudowyj. Konserwacja indeksów podczas wstawiania zbiorczego zwiększa narzut.
• Użyj CheckConstraints: false (domyślnie) w przypadku zaufanych danych, aby pominąć sprawdzanie ograniczeń podczas kopiowania zbiorczego.

Limitations

• Kopia masowa nie obsługuje kolumn chronionych przez Always Encrypted. Aby uzyskać więcej informacji, zobacz Ograniczenia.
• Kopiowanie masowe na poziomie TDS nie jest obsługiwane w Azure SQL Database. W przypadku Azure SQL Database zamiast tego użyj wsadowych instrukcji INSERT lub wzorca tabeli przejściowej.

Treści powiązane

• Dostrajanie wydajności
• Zapytania i oświadczenia
• Mapowania typów danych
• Przewodnik migracji
• Obsługa błędów i wzorce powtórek