Hromadné operace s go-mssqldb

Ovladač go-mssqldb podporuje vysoce výkonné operace hromadného vkládání pomocí funkce mssql.CopyIn. Hromadné vkládání obchází běžnou cestu po jednotlivých řádcích INSERT a přenáší data přímo na server pomocí protokolu TDS pro hromadné kopírování.

Vyberte hromadnou kopii, TVP nebo JSON

Použijte následující návod, když potřebujete poslat více řádků nebo složité payloady do SQL Server:

Zvolte...	Když to nejlépe sedí	Kompromis
Hromadné kopírování pomocí mssql.CopyIn	Potřebujete nejrychlejší způsob, jak načíst více řádků do jedné cílové tabulky.	Nejlepší propustnost, ale je zaměřena na načítání dat do tabulek spíše než na rozhraní uložených procedur nebo datové struktury různých tvarů.
Parametr s tabulkovou hodnotou	Musíte předat striktně typovanou sadu řádků dat do uložené procedury nebo parametrizovaného příkazu.	Zachovává hranice schématu a procedur, ale vyžaduje uživatelem definovaný typ tabulky a odpovídající pořadí polí.
JSON s OPENJSON nebo FOR JSON	Vaše aplikace už používá JSON nebo je struktura přenášených dat vnořená či proměnlivá.	Pro kód aplikace je přenositelnější, ale pro strukturované vkládání bývá obvykle pomalejší a méně typově bezpečný než TVP nebo hromadné kopírování dat.

Pokud načítáte velké dávky do přechodné nebo cílové tabulky, začněte hromadným kopírováním. Pokud voláte uložené procedury pomocí strukturovaných sad řádků, začněte s parametry TVP. Pokud potřebujete vnořené dokumenty nebo volná schémata, začněte s JSON.

Příklady v tomto článku jsou porovnány s databází AdventureWorks2025 . Příklady hromadného kopírování jsou určeny pro HumanResources.Department a Production.ProductCategory.

Základní hromadná vložka

Použijte mssql.CopyIn k vytvoření hromadného kopírovacího příkazu a poté použijte Exec pro posílání řádků:

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

Konečné stmt.Exec() volání bez argumentů vyčistí všechny zbývající řádky a dokončí hromadnou kopírovací operaci.

BulkOptions

Struktura mssql.BulkOptions konfiguruje chování hromadného kopírovaní:

Obor	Typ	Description
CheckConstraints	bool	Zkontrolujte omezení během hromadného vložení.
FireTriggers	bool	Spusťte INSERT triggery na cílové tabulce.
KeepNulls	bool	Zachovejte nulové hodnoty místo vkládání výchozích hodnot.
KilobytesPerBatch	int	Kilobajty na dávku. 0 používá výchozí nastavení serveru.
RowsPerBatch	int	Řady na várku. 0 používá výchozí nastavení serveru.
Order	[]string	ORDER direktiva pro cílový clusterovaný index (například []string{"Id ASC"}).
Tablock	bool	Získejte zámek na úrovni tabulky po dobu trvání hromadné kopie.

Příklad s možnostmi

Předejte BulkOptions, chcete-li řídit kontroly omezení, triggery a zamykání:

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

Zpracování chyb

Pokud nějaký řádek selže, celá operace hromadného kopírování selže. Zkontrolujte chyby jak při volání každého řádku Exec, tak při finálním vyprázdnění 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
}

Detekce a zaznamenávání neúspěšných řádků

Když hromadná kopírovací operace selže, chybová zpráva ze SQL Server indikuje omezení nebo datový problém, ale neidentifikuje konkrétní řádek. Pro identifikaci neúspěšných řádků použijte metodu dávkování:

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

Pokud potřebujete přeskočit chybné řádky a pokračovat, použijte jednotlivé příkazy INSERT nebo postup se staging tabulkou: hromadně zkopírujte data do staging tabulky bez omezení a pak použijte MERGE nebo INSERT...SELECT s ošetřením chyb pro přesun dat do cílové tabulky.

Stream z CSV souborů

U velkých CSV souborů streamujte řádky přímo ze souboru do hromadné kopie bez načítání celého souboru do paměti:

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

Porovnání výkonu

Hromadné kopírování je výrazně rychlejší než jednotlivé vklady pro velké datové zatížení. Následující tabulka ukazuje přibližné výkonnostní charakteristiky pro vložení 100 000 řádků:

Method	Relativní rychlost	Zpáteční lety sítě	Uzamčení
Individuální INSERT	Nejpomalejší (1x)	100 000	Úroveň řádků na každé vložení.
Dávkované INSERT (1 000 řádků v jednom příkazu)	Střední (5-10x)	100	Úroveň řádků na várku.
Hromadné kopírování bez TABLOCK	Rychlé (20–50×)	Záleží na velikosti dávky	Hromadně na úrovni řádků.
Hromadné kopírování s TABLOCK	Nejrychlejší (50-100x)	Záleží na velikosti dávky	Zámek na úrovni tabulky, minimální protokolování.

Note

Skutečný výkon se liší v závislosti na latenci sítě, konfiguraci serveru, indexech tabulek a na tom, zda je k dispozici minimální logování. Porovnávejte s vaším konkrétním pracovním zatížením pomocí testing.B. Viz Ladění výkonu.

Pořadí sloupců a mapování typů

Sloupce v mssql.CopyIn musí odpovídat pořadí a typům, které cílová tabulka očekává. Ovladač neprovádí porovnání názvů sloupců; používá pozicové mapování.

Běžné typy

Typ Go	sloupec serveru SQL Server	Issue	Solution
string	varchar	Implicitní nvarchar konverze.	Použijte mssql.VarChar obal.
float64	decimal(18,4)	Ztráta přesnosti.	Vystupovat jako string.
time.Time	datetime2	Převod časových pásem.	Používejte UTC časy.
nil	Libovolný nulovatelný sloupec	Vyžaduje KeepNulls: true.	Nastavit KeepNulls v BulkOptions.

Příklad s explicitními typy

Specifikujte typy sloupců explicitně, když výchozí mapování typů neodpovídá vašemu schématu:

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

Tipy týkající se výkonu

• Použijte Tablock pro hromadné vkládání do prázdných tabulek. Tato možnost snižuje spory o zámky a umožňuje minimální logování.
• Nastavte RowsPerBatch, abyste určili, jak často ovladač odesílá data. Větší dávky snižují počet cest tam a zpět, ale spotřebovávají více paměti.
• Zvyšte packet size v připojovacím řetězci (až na 32767), aby se snížila režie sítě.
• Seřaďte data tak, aby odpovídala seskupenému indexu cílové tabulky, a nastavte možnost Order. Tento přístup se vyhýbá třídění na straně serveru.
• Odstraňte neclusterované indexy před rozsáhlým hromadným načítáním dat a poté je znovu sestavte. Údržba indexů při hromadném vkládání zvyšuje režii.
• Použijte CheckConstraints: false (výchozí možnost) pro důvěryhodná data, aby se při hromadném kopírování přeskočila kontrola omezení.

Omezení

• Hromadné kopírování nepodporuje sloupce chráněné systémem Always Encrypted. Další informace najdete v tématu Omezení.
• Hromadné kopírovaní na úrovni TDS není v Azure SQL Database podporováno. Pro Azure SQL Database použijte místo toho batched INSERT příkazy nebo vzor staging-table.

Související obsah

• Ladění výkonu
• Dotazy a prohlášení
• Mapování datových typů
• Průvodce migrací
• Zpracování chyb a vzory opakovaných pokusů