Import / Export DataImport / Export Data

dòng trong một file importrows in one import file

1M

dòng trong một file importrows in one import file

cho 1 triệu dòngfor a million rows

< 5 min

cho 1 triệu dòngfor a million rows

heap tối đa của workermax worker heap

512 MB

heap tối đa của workermax worker heap

dòng mỗi lần exportrows per export

500k

dòng mỗi lần exportrows per export

File được tải thẳng lên S3, API chỉ tạo job và trả 202. Reader đọc file theo stream, cắt thành lô 1.000 dòng và đẩy vào hàng đợi; worker pool kiểm tra, ghi lỗi và upsert vào Postgres, rồi gom kết quả khi lô cuối xong. Export đọc cursor trên replica và stream lên S3. Bật “2% dòng lỗi” hoặc “Export 500k dòng” để xem hai luồng còn lại.The file is uploaded straight to S3; the API only creates a job and answers 202. The reader streams the file, cuts it into 1,000-row chunks and queues them; a worker pool validates, records errors and upserts into Postgres, and the results are gathered once the last chunk is done. Exports read a cursor on a replica and stream to S3. Turn on “2% invalid rows” or “Export 500k rows” to see the other two flows.

1. 1

   OutOfMemory khi đọc cả fileOutOfMemory from loading the whole file

   Apache POI XSSFWorkbook dựng cả file Excel trong RAM, 1 triệu dòng có thể cần vài GB. Dùng parser stream (univocity, encoding/csv, POI SAX/excel-streaming-reader, excelize Rows) và chỉ giữ một lô trong bộ nhớ.Apache POI's XSSFWorkbook builds the whole workbook in RAM; a million rows can need several GB. Use a streaming parser (univocity, encoding/csv, POI SAX/excel-streaming-reader, excelize Rows) and keep one chunk in memory.

   IteratorPipeline
2. 2

   Insert từng dòng quá chậmRow-by-row inserts are too slow

   1 triệu INSERT là 1 triệu round trip. Batch 1.000 dòng với reWriteBatchedInserts=true đạt vài chục nghìn dòng/giây; COPY vào bảng tạm rồi INSERT … SELECT ON CONFLICT còn nhanh hơn. Thêm worker pool 8 luồng, giới hạn theo pool connection.A million INSERTs are a million round trips. Batches of 1,000 with reWriteBatchedInserts=true reach tens of thousands of rows a second; COPY into a temp table followed by INSERT … SELECT ON CONFLICT is faster still. Add an 8-worker pool, capped by the connection pool.

   Worker PoolFan-out / Fan-in
3. 3

   Một dòng sai làm hỏng cả fileOne bad row fails the whole file

   Kiểm tra từng dòng trước khi ghi, gom lỗi theo dòng và cột vào import_error, dòng đúng vẫn ghi. Đặt ngưỡng: lỗi quá 10% thì dừng job để tránh nạp một file sai cột.Validate each row before writing, collect errors per line and column in import_error, and still write the good rows. Set a threshold: above 10% errors the job stops, which catches a file with shifted columns.

   PipelineCommand
4. 4

   Chạy lại tạo bản ghi trùngRe-runs create duplicates

   Kafka giao ít nhất một lần, admin cũng có thể import lại cùng file. Bảng import_chunk unique(job_id, chunk_no) chặn lô đã xử lý; upsert theo khoá tự nhiên external_id làm import lại cùng dữ liệu không đổi kết quả.Kafka delivers at least once and admins may re-import the same file. import_chunk with unique(job_id, chunk_no) blocks finished chunks; upserting on the natural key external_id makes a repeated import leave the result unchanged.

   CommandProducer-Consumer
5. 5

   Export làm treo request và chậm DB chínhExports hang requests and slow the primary

   Không build file trong request HTTP. Chạy job nền, đọc server-side cursor trên replica (fetchSize 5.000, autocommit tắt với pgjdbc), stream qua gzip vào S3 multipart rồi gửi link. OFFSET phân trang sâu chậm dần, cursor hay keyset thì không.Never build the file inside the HTTP request. Run a background job that reads a server-side cursor on a replica (fetchSize 5,000, autocommit off with pgjdbc), streams through gzip into an S3 multipart upload and sends a link. Deep OFFSET paging slows down; a cursor or keyset does not.

   IteratorProducer-Consumer
6. 6

   Nhiều định dạng nguồnMany source formats

   Adapter đưa CSV, XLSX và JSON Lines về cùng một Iterator<Row>; phần còn lại của pipeline không biết file gốc là gì. Facade ImportService che các bước upload, tạo job, theo dõi.Adapters turn CSV, XLSX and JSON Lines into the same Iterator<Row>; the rest of the pipeline never knows the source format. An ImportService facade hides upload, job creation and tracking.

   AdapterFacadeIterator

Chunk worker: nhận một lô 1.000 dòng, bỏ qua nếu lô đã xử lý, kiểm tra từng dòng, ghi dòng đúng theo lô, lưu lỗi và cộng tiến độ trong cùng một transaction.The chunk worker takes one 1,000-row chunk, skips it if already done, validates every row, writes the good rows as a batch, stores the errors and bumps progress, all in one transaction.

@Component
@RequiredArgsConstructor
public class ImportChunkWorker {
    private final JdbcTemplate jdbc;
    private final RowValidator validator;
    private final ImportErrorRepository errors;
    private final ImportJobRepository jobs;

    @KafkaListener(topics = "import.chunks", concurrency = "8")1
    @Transactional
    public void handle(ChunkMessage m) {
        int claimed = jdbc.update("""
            INSERT INTO import_chunk(job_id, chunk_no) VALUES (?, ?)
            ON CONFLICT DO NOTHING""", m.jobId(), m.chunkNo());
        if (claimed == 0) return;                          // chunk already done2

        List<CustomerRow> valid = new ArrayList<>(m.rows().size());
        List<ImportError> bad = new ArrayList<>();
        for (CsvRow r : m.rows()) {
            List<FieldError> problems = validator.check(r);
            if (problems.isEmpty()) valid.add(CustomerRow.from(r));
            else problems.forEach(p -> bad.add(ImportError.of(m.jobId(), r.lineNo(), p)));3
        }

        jdbc.batchUpdate("""
            INSERT INTO customer(external_id, email, name, updated_at)
            VALUES (?, ?, ?, now())
            ON CONFLICT (external_id) DO UPDATE
              SET email = EXCLUDED.email, name = EXCLUDED.name, updated_at = now()""",
            valid, 1000, (ps, c) -> {4
                ps.setString(1, c.externalId());
                ps.setString(2, c.email());
                ps.setString(3, c.name());
            });
        errors.saveAll(bad);
        jobs.addProgress(m.jobId(), m.rows().size(), bad.size()); // done_chunks + 1
    }
}

func (w *ChunkWorker) Handle(ctx context.Context, m ChunkMessage) error {
	return pgx.BeginFunc(ctx, w.db, func(tx pgx.Tx) error {1
		tag, err := tx.Exec(ctx, `INSERT INTO import_chunk(job_id, chunk_no)
			VALUES ($1, $2) ON CONFLICT DO NOTHING`, m.JobID, m.ChunkNo)
		if err != nil || tag.RowsAffected() == 0 {
			return err // nil when the chunk is already done
		}

		var valid, bad [][]any
		for _, r := range m.Rows {
			if probs := w.validate(r); len(probs) > 0 {
				for _, p := range probs {
					bad = append(bad, []any{m.JobID, r.Line, p.Column, p.Code})
				}
				continue
			}
			valid = append(valid, []any{r.ExternalID, r.Email, r.Name})
		}

		if _, err := tx.Exec(ctx, `CREATE TEMP TABLE stage
			(external_id text, email text, name text) ON COMMIT DROP`); err != nil {
			return err
		}
		cols := []string{"external_id", "email", "name"}
		if _, err := tx.CopyFrom(ctx, pgx.Identifier{"stage"}, cols, pgx.CopyFromRows(valid)); err != nil {2
			return err
		}
		if _, err := tx.Exec(ctx, `INSERT INTO customer(external_id, email, name)3
			SELECT external_id, email, name FROM stage
			ON CONFLICT (external_id) DO UPDATE
			SET email = EXCLUDED.email, name = EXCLUDED.name`); err != nil {
			return err
		}
		errCols := []string{"job_id", "line_no", "column_name", "code"}
		if _, err := tx.CopyFrom(ctx, pgx.Identifier{"import_error"}, errCols, pgx.CopyFromRows(bad)); err != nil {
			return err
		}
		_, err = tx.Exec(ctx, `UPDATE import_job SET processed = processed + $2,
			failed = failed + $3, done_chunks = done_chunks + 1 WHERE id = $1`,4
			m.JobID, len(m.Rows), len(bad))
		return err
	})
}

Mỗi pattern mở bài học tương ứng trong menu Pattern.Each one opens its lesson in the Patterns menu.