1時間30分かかっていたデータ取り込み処理をたった5分で終わらせる技術〜ISUCONは役にたつ〜

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/gocarina/gocsv"
	"github.com/jmoiron/sqlx"
)

func main() {
	ctx := context.Background()

	// MySQLコンテナを起動
	connStr, cleanup, err := startMySQLContainer(ctx)
	if err != nil {
		log.Fatal(err)
	}
	defer cleanup()

	db, err := sqlx.Open("mysql", connStr)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	// テーブル・マスターデータを作成
	if err := setupTables(db); err != nil {
		log.Fatal(err)
	}

	// サンプルCSVを生成（50万行）
	csvFilename := "data.csv"
	if err := generateSampleCSV(csvFilename, 500000); err != nil {
		log.Fatal(err)
	}

	// 1. CSVを読み取る
	file, err := os.Open(csvFilename)
	if err != nil {
		log.Fatal(err)
	}
	defer file.Close()

	var products []Product
	if err := gocsv.UnmarshalFile(file, &products); err != nil {
		log.Fatal(err)
	}
	fmt.Printf("CSV読み込み完了: %d 行\n", len(products))

	importStart := time.Now()

	for i, product := range products {
		// 2. 読んでない行があれば1行読み取る、なければ終了
		lineNum := i + 2

		// 3. category_codeをcategory_idに変換
		var category Category
		if err := db.Get(
			&category,
			`SELECT * FROM categories WHERE code = ?`,
			product.CategoryCode,
		); err != nil {
			log.Fatalf("行 %d: category_code %q の検索に失敗: %v", lineNum, product.CategoryCode, err)
		}

		// 4. supplier_codeをsupplier_idに変換
		var supplier Supplier
		if err := db.Get(
			&supplier,
			`SELECT * FROM suppliers WHERE code = ?`,
			product.SupplierCode,
		); err != nil {
			log.Fatalf("行 %d: supplier_code %q の検索に失敗: %v", lineNum, product.SupplierCode, err)
		}

		// 5. status_codeをstatus_idに変換
		var status Status
		if err := db.Get(
			&status,
			`SELECT * FROM statuses WHERE code = ?`,
			product.StatusCode,
		); err != nil {
			log.Fatalf("行 %d: status_code %q の検索に失敗: %v", lineNum, product.StatusCode, err)
		}

		// 6. ProductRowに変換
		row := ProductRow{
			ProductCode: product.ProductCode,
			ProductName: product.ProductName,
			CategoryID:  category.ID,
			SupplierID:  supplier.ID,
			StatusID:    status.ID,
			UnitPrice:   product.UnitPrice,
		}

		// 7. UPDATE文を実行する
		result, err := db.NamedExec(`
			UPDATE products
			SET product_name = :product_name,
				category_id = :category_id,
				supplier_id = :supplier_id,
				status_id = :status_id,
				unit_price = :unit_price
			WHERE product_code = :product_code`,
			row,
		)
		if err != nil {
			log.Fatalf("行 %d: productsの更新に失敗: %v", lineNum, err)
		}

		rowsAffected, err := result.RowsAffected()
		if err != nil {
			log.Fatalf("行 %d: 更新件数の取得に失敗: %v", lineNum, err)
		}

		// 8. UPDATE対象がなければINSERTする
		if rowsAffected == 0 {
			_, err = db.NamedExec(`
				INSERT INTO products (product_code, product_name, category_id, supplier_id, status_id, unit_price)
				VALUES (:product_code, :product_name, :category_id, :supplier_id, :status_id, :unit_price)`,
				row,
			)
			if err != nil {
				log.Fatalf("行 %d: productsの登録に失敗: %v", lineNum, err)
			}
		}

		if (lineNum-1)%1000 == 0 {
			rate := float64(lineNum-1) / time.Since(importStart).Seconds()
			fmt.Printf("進捗: %d / %d 行 (%.0f 行/秒)\n", lineNum-1, len(products), rate)
		}
		// 9. 2に戻る
	}

	fmt.Printf("完了: %d 行 (所要時間: %v)\n", len(products), time.Since(importStart))
}

package main

type Product struct {
	ProductCode  string `csv:"product_code"`
	ProductName  string `csv:"product_name"`
	CategoryCode string `csv:"category_code"`
	SupplierCode string `csv:"supplier_code"`
	StatusCode   string `csv:"status_code"`
	UnitPrice    int    `csv:"unit_price"`
}

type Category struct {
	ID   int    `db:"id"`
	Code string `db:"code"`
	Name string `db:"name"`
}

type Supplier struct {
	ID   int    `db:"id"`
	Code string `db:"code"`
	Name string `db:"name"`
}

type Status struct {
	ID   int    `db:"id"`
	Code string `db:"code"`
	Name string `db:"name"`
}

type ProductRow struct {
	ProductCode string `db:"product_code"`
	ProductName string `db:"product_name"`
	CategoryID  int    `db:"category_id"`
	SupplierID  int    `db:"supplier_id"`
	StatusID    int    `db:"status_id"`
	UnitPrice   int    `db:"unit_price"`
}

package main

import (
	"context"
	"encoding/csv"
	"fmt"
	"math/rand"
	"os"
	"strconv"
	"time"

	"github.com/jmoiron/sqlx"
	"github.com/testcontainers/testcontainers-go"
	"github.com/testcontainers/testcontainers-go/modules/mysql"
	"github.com/testcontainers/testcontainers-go/wait"
)

func startMySQLContainer(ctx context.Context) (connStr string, cleanup func(), err error) {
	mysqlContainer, err := mysql.Run(ctx,
		"mysql:8.0",
		mysql.WithDatabase("testdb"),
		mysql.WithUsername("user"),
		mysql.WithPassword("password"),
		testcontainers.WithWaitStrategyAndDeadline(3*time.Minute,
			wait.ForListeningPort("3306/tcp").
				WithStartupTimeout(3*time.Minute),
		),
	)
	if err != nil {
		return "", nil, err
	}

	connStr, err = mysqlContainer.ConnectionString(ctx)
	if err != nil {
		_ = mysqlContainer.Terminate(ctx)
		return "", nil, err
	}

	cleanup = func() {
		_ = mysqlContainer.Terminate(ctx)
	}
	return connStr, cleanup, nil
}

func generateSampleCSV(filename string, rows int) error {
	file, err := os.Create(filename)
	if err != nil {
		return err
	}
	defer file.Close()

	writer := csv.NewWriter(file)
	defer writer.Flush()

	if err := writer.Write([]string{"product_code", "product_name", "category_code", "supplier_code", "status_code", "unit_price"}); err != nil {
		return err
	}

	categoryCodes := []string{"CAT01", "CAT02", "CAT03"}
	supplierCodes := []string{"SUP01", "SUP02", "SUP03"}
	statusCodes := []string{"active", "discontinued", "pending"}

	for i := 0; i < rows; i++ {
		record := []string{
			fmt.Sprintf("P%d", 1000+i+1),
			fmt.Sprintf("商品_%d", i+1),
			categoryCodes[rand.Intn(len(categoryCodes))],
			supplierCodes[rand.Intn(len(supplierCodes))],
			statusCodes[rand.Intn(len(statusCodes))],
			strconv.Itoa(rand.Intn(10000) + 100),
		}
		if err := writer.Write(record); err != nil {
			return err
		}
	}

	return nil
}

func setupTables(db *sqlx.DB) error {
	tables := []string{
		`CREATE TABLE IF NOT EXISTS categories (
			id INT AUTO_INCREMENT PRIMARY KEY,
			code VARCHAR(10) UNIQUE NOT NULL,
			name VARCHAR(100) NOT NULL
		)`,
		`CREATE TABLE IF NOT EXISTS suppliers (
			id INT AUTO_INCREMENT PRIMARY KEY,
			code VARCHAR(10) UNIQUE NOT NULL,
			name VARCHAR(100) NOT NULL
		)`,
		`CREATE TABLE IF NOT EXISTS statuses (
			id INT AUTO_INCREMENT PRIMARY KEY,
			code VARCHAR(20) UNIQUE NOT NULL,
			name VARCHAR(100) NOT NULL
		)`,
		`CREATE TABLE IF NOT EXISTS products (
			id INT AUTO_INCREMENT PRIMARY KEY,
			product_code VARCHAR(50) UNIQUE NOT NULL,
			product_name VARCHAR(255) NOT NULL,
			category_id INT NOT NULL,
			supplier_id INT NOT NULL,
			status_id INT NOT NULL,
			unit_price INT NOT NULL,
			FOREIGN KEY (category_id) REFERENCES categories(id),
			FOREIGN KEY (supplier_id) REFERENCES suppliers(id),
			FOREIGN KEY (status_id) REFERENCES statuses(id)
		)`,
	}

	for _, table := range tables {
		if _, err := db.Exec(table); err != nil {
			return err
		}
	}

	masterData := []string{
		`INSERT IGNORE INTO categories (code, name) VALUES
			('CAT01', '文房具'), ('CAT02', '食品'), ('CAT03', '電化製品')`,
		`INSERT IGNORE INTO suppliers (code, name) VALUES
			('SUP01', '株式会社A商事'), ('SUP02', '株式会社B産業'), ('SUP03', '株式会社C物産')`,
		`INSERT IGNORE INTO statuses (code, name) VALUES
			('active', '販売中'), ('discontinued', '販売終了'), ('pending', '販売準備中')`,
	}

	for _, data := range masterData {
		if _, err := db.Exec(data); err != nil {
			return err
		}
	}

	return nil
}

module csv-import-example

go 1.24.5

require (
	github.com/go-sql-driver/mysql v1.9.3
	github.com/gocarina/gocsv v0.0.0-20240520201108-78e41c74b4b1
	github.com/jmoiron/sqlx v1.4.0
	github.com/testcontainers/testcontainers-go v0.40.0
	github.com/testcontainers/testcontainers-go/modules/mysql v0.40.0
)

require (
	dario.cat/mergo v1.0.2 // indirect
	filippo.io/edwards25519 v1.1.0 // indirect
	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
	github.com/Microsoft/go-winio v0.6.2 // indirect
	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
	github.com/containerd/errdefs v1.0.0 // indirect
	github.com/containerd/errdefs/pkg v0.3.0 // indirect
	github.com/containerd/log v0.1.0 // indirect
	github.com/containerd/platforms v0.2.1 // indirect
	github.com/cpuguy83/dockercfg v0.3.2 // indirect
	github.com/davecgh/go-spew v1.1.1 // indirect
	github.com/distribution/reference v0.6.0 // indirect
	github.com/docker/docker v28.5.1+incompatible // indirect
	github.com/docker/go-connections v0.6.0 // indirect
	github.com/docker/go-units v0.5.0 // indirect
	github.com/ebitengine/purego v0.8.4 // indirect
	github.com/felixge/httpsnoop v1.0.4 // indirect
	github.com/go-logr/logr v1.4.3 // indirect
	github.com/go-logr/stdr v1.2.2 // indirect
	github.com/go-ole/go-ole v1.2.6 // indirect
	github.com/google/uuid v1.6.0 // indirect
	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7 // indirect
	github.com/klauspost/compress v1.18.0 // indirect
	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
	github.com/magiconair/properties v1.8.10 // indirect
	github.com/moby/docker-image-spec v1.3.1 // indirect
	github.com/moby/go-archive v0.1.0 // indirect
	github.com/moby/patternmatcher v0.6.0 // indirect
	github.com/moby/sys/sequential v0.6.0 // indirect
	github.com/moby/sys/user v0.4.0 // indirect
	github.com/moby/sys/userns v0.1.0 // indirect
	github.com/moby/term v0.5.0 // indirect
	github.com/morikuni/aec v1.0.0 // indirect
	github.com/opencontainers/go-digest v1.0.0 // indirect
	github.com/opencontainers/image-spec v1.1.1 // indirect
	github.com/pkg/errors v0.9.1 // indirect
	github.com/pmezard/go-difflib v1.0.0 // indirect
	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
	github.com/shirou/gopsutil/v4 v4.25.6 // indirect
	github.com/sirupsen/logrus v1.9.3 // indirect
	github.com/stretchr/testify v1.11.1 // indirect
	github.com/tklauser/go-sysconf v0.3.12 // indirect
	github.com/tklauser/numcpus v0.6.1 // indirect
	github.com/yusufpapurcu/wmi v1.2.4 // indirect
	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
	go.opentelemetry.io/otel v1.38.0 // indirect
	go.opentelemetry.io/otel/metric v1.38.0 // indirect
	go.opentelemetry.io/otel/sdk v1.38.0 // indirect
	go.opentelemetry.io/otel/trace v1.38.0 // indirect
	golang.org/x/crypto v0.43.0 // indirect
	golang.org/x/sys v0.38.0 // indirect
	google.golang.org/grpc v1.78.0 // indirect
	google.golang.org/protobuf v1.36.11 // indirect
	gopkg.in/yaml.v3 v3.0.1 // indirect
)

diff --git a/main.go b/main.go
index c3705d8..c3c16cf 100644
--- a/main.go
+++ b/main.go
@@ -52,6 +52,34 @@ func main() {
 	}
 	fmt.Printf("CSV読み込み完了: %d 行\n", len(products))

+	// マスターデータをmapに読み込み（code → id）
+	var categories []Category
+	if err := db.Select(&categories, "SELECT * FROM categories"); err != nil {
+		log.Fatal(err)
+	}
+	categoryMap := make(map[string]int, len(categories))
+	for _, c := range categories {
+		categoryMap[c.Code] = c.ID
+	}
+
+	var suppliers []Supplier
+	if err := db.Select(&suppliers, "SELECT * FROM suppliers"); err != nil {
+		log.Fatal(err)
+	}
+	supplierMap := make(map[string]int, len(suppliers))
+	for _, s := range suppliers {
+		supplierMap[s.Code] = s.ID
+	}
+
+	var statuses []Status
+	if err := db.Select(&statuses, "SELECT * FROM statuses"); err != nil {
+		log.Fatal(err)
+	}
+	statusMap := make(map[string]int, len(statuses))
+	for _, s := range statuses {
+		statusMap[s.Code] = s.ID
+	}
+
 	importStart := time.Now()

 	for i, product := range products {
@@ -59,41 +87,29 @@ func main() {
 		lineNum := i + 2

 		// 3. category_codeをcategory_idに変換
-		var category Category
-		if err := db.Get(
-			&category,
-			`SELECT * FROM categories WHERE code = ?`,
-			product.CategoryCode,
-		); err != nil {
-			log.Fatalf("行 %d: category_code %q の検索に失敗: %v", lineNum, product.CategoryCode, err)
+		categoryID, ok := categoryMap[product.CategoryCode]
+		if !ok {
+			log.Fatalf("行 %d: category_code %q の検索に失敗", lineNum, product.CategoryCode)
 		}

 		// 4. supplier_codeをsupplier_idに変換
-		var supplier Supplier
-		if err := db.Get(
-			&supplier,
-			`SELECT * FROM suppliers WHERE code = ?`,
-			product.SupplierCode,
-		); err != nil {
-			log.Fatalf("行 %d: supplier_code %q の検索に失敗: %v", lineNum, product.SupplierCode, err)
+		supplierID, ok := supplierMap[product.SupplierCode]
+		if !ok {
+			log.Fatalf("行 %d: supplier_code %q の検索に失敗", lineNum, product.SupplierCode)
 		}

 		// 5. status_codeをstatus_idに変換
-		var status Status
-		if err := db.Get(
-			&status,
-			`SELECT * FROM statuses WHERE code = ?`,
-			product.StatusCode,
-		); err != nil {
-			log.Fatalf("行 %d: status_code %q の検索に失敗: %v", lineNum, product.StatusCode, err)
+		statusID, ok := statusMap[product.StatusCode]
+		if !ok {
+			log.Fatalf("行 %d: status_code %q の検索に失敗", lineNum, product.StatusCode)
 		}

 		row := ProductRow{
 			ProductCode: product.ProductCode,
 			ProductName: product.ProductName,
-			CategoryID:  category.ID,
-			SupplierID:  supplier.ID,
-			StatusID:    status.ID,
+			CategoryID:  categoryID,
+			SupplierID:  supplierID,
+			StatusID:    statusID,
 			UnitPrice:   product.UnitPrice,
 		}

diff --git a/main.go b/main.go
index c3c16cf..0da4db0 100644
--- a/main.go
+++ b/main.go
@@ -113,36 +113,23 @@ func main() {
 			UnitPrice:   product.UnitPrice,
 		}

-		// 7. UPDATE文を実行する
-		result, err := db.NamedExec(`
-			UPDATE products
-			SET product_name = :product_name,
-				category_id = :category_id,
-				supplier_id = :supplier_id,
-				status_id = :status_id,
-				unit_price = :unit_price
-			WHERE product_code = :product_code`,
+		// 7. UPSERT（INSERT or UPDATE）を実行する
+		_, err := db.NamedExec(`
+			INSERT INTO products (
+				product_code, product_name, category_id, supplier_id, status_id, unit_price
+			) VALUES (
+				:product_code, :product_name, :category_id, :supplier_id, :status_id, :unit_price
+			)
+			ON DUPLICATE KEY UPDATE
+				product_name = VALUES(product_name),
+				category_id  = VALUES(category_id),
+				supplier_id  = VALUES(supplier_id),
+				status_id    = VALUES(status_id),
+				unit_price   = VALUES(unit_price)`,
 			row,
 		)
 		if err != nil {
-			log.Fatalf("行 %d: productsの更新に失敗: %v", lineNum, err)
-		}
-
-		rowsAffected, err := result.RowsAffected()
-		if err != nil {
-			log.Fatalf("行 %d: 更新件数の取得に失敗: %v", lineNum, err)
-		}
-
-		// 8. UPDATE対象がなければINSERTする
-		if rowsAffected == 0 {
-			_, err = db.NamedExec(`
-				INSERT INTO products (product_code, product_name, category_id, supplier_id, status_id, unit_price)
-				VALUES (:product_code, :product_name, :category_id, :supplier_id, :status_id, :unit_price)`,
-				row,
-			)
-			if err != nil {
-				log.Fatalf("行 %d: productsの登録に失敗: %v", lineNum, err)
-			}
+			log.Fatalf("行 %d: productsのUPSERTに失敗: %v", lineNum, err)
 		}

 		if (lineNum-1)%1000 == 0 {

diff --git a/main.go b/main.go
index 0da4db0..daf2689 100644
--- a/main.go
+++ b/main.go
@@ -80,8 +80,8 @@ func main() {
 		statusMap[s.Code] = s.ID
 	}

-	importStart := time.Now()
-
+	// code → id 変換してProductRowスライスを構築
+	var rows []ProductRow
 	for i, product := range products {
 		// 2. 読んでない行があれば1行読み取る、なければ終了
 		lineNum := i + 2
@@ -104,16 +104,29 @@ func main() {
 			log.Fatalf("行 %d: status_code %q の検索に失敗", lineNum, product.StatusCode)
 		}

-		row := ProductRow{
+		// 6. ProductRowに変換
+		rows = append(rows, ProductRow{
 			ProductCode: product.ProductCode,
 			ProductName: product.ProductName,
 			CategoryID:  categoryID,
 			SupplierID:  supplierID,
 			StatusID:    statusID,
 			UnitPrice:   product.UnitPrice,
+		})
+	}
+	fmt.Printf("変換完了: %d 行\n", len(rows))
+
+	// バルクUPSERT（1000行ずつ）
+	const batchSize = 1000
+	importStart := time.Now()
+
+	for i := 0; i < len(rows); i += batchSize {
+		end := i + batchSize
+		if end > len(rows) {
+			end = len(rows)
 		}
+		batch := rows[i:end]

-		// 6. UPSERT（INSERT or UPDATE）を実行する
 		_, err := db.NamedExec(`
 			INSERT INTO products (
 				product_code, product_name, category_id, supplier_id, status_id, unit_price
@@ -126,17 +139,16 @@ func main() {
 				supplier_id  = VALUES(supplier_id),
 				status_id    = VALUES(status_id),
 				unit_price   = VALUES(unit_price)`,
-			row,
+			batch,
 		)
 		if err != nil {
-			log.Fatalf("行 %d: productsのUPSERTに失敗: %v", lineNum, err)
+			log.Fatalf("バッチ %d-%d: UPSERTに失敗: %v", i+1, end, err)
 		}

-		if (lineNum-1)%1000 == 0 {
-			rate := float64(lineNum-1) / time.Since(importStart).Seconds()
-			fmt.Printf("進捗: %d / %d 行 (%.0f 行/秒)\n", lineNum-1, len(products), rate)
+		if end%10000 == 0 || end == len(rows) {
+			rate := float64(end) / time.Since(importStart).Seconds()
+			fmt.Printf("進捗: %d / %d 行 (%.0f 行/秒)\n", end, len(rows), rate)
 		}
-		// 8. 2に戻る
 	}

 	fmt.Printf("完了: %d 行 (所要時間: %v)\n", len(products), time.Since(importStart))