본문으로 바로가기 본문 바로가기
WIKI 기술 지식 베이스

REST API로 D1에 대량 가져오기(Bulk import)

원문 보기 위키 갱신

이 튜토리얼에서는 REST API를 사용해서 데이터베이스를 D1로 가져오는 방법을 배워 볼게요.

출처: 문서

본문

사전 준비 (Prerequisites)

  1. Cloudflare 계정 ↗︎에 가입하세요.
  2. Node.js ↗︎를 설치하세요.

Node.js 버전 관리자

Volta ↗︎나 nvm ↗︎ 같은 Node 버전 관리자를 사용하면 권한 문제를 피하고 Node.js 버전을 쉽게 바꿀 수 있어요. 이 가이드에서 뒤에 설명할 Wrangler는 Node 16.17.0 이상 버전을 요구해요.

1. D1 API 토큰 만들기

REST API를 사용하려면 API 요청을 인증할 API 토큰을 생성해야 해요. Cloudflare 대시보드를 통해 생성할 수 있어요.

  1. Cloudflare 대시보드에서 API Tokens 페이지로 이동하세요. Account API tokens ↗로 이동
  2. API Tokens 아래에서 Create Token을 선택하세요.
  3. Custom token > Create custom token으로 스크롤한 후 Get started를 선택하세요.
  4. Token name 아래에 설명적인 토큰 이름을 입력하세요. 예를 들어 Name-D1-Import-API-Token처럼요.
  5. Permissions 아래에서:
    • Account를 선택하세요.
    • D1을 선택하세요.
    • Edit를 선택하세요.
  6. Continue to summary를 선택하세요.
  7. Create token을 선택하세요.
  8. API 토큰을 복사해서 안전한 파일에 저장하세요.
  • Cloudflare 대시보드에서 API 토큰을 만드는 방법에 대한 자세한 내용은 API 토큰 만들기를 참고하세요.
  • API를 통해 토큰을 만드는 방법에 대한 자세한 내용은 API로 토큰 만들기를 참고하세요.

2. 대상 테이블 만들기

가져오려는 데이터의 스키마와 일치하는 기존 D1 테이블이 있어야 해요.

이 튜토리얼에서는 다음을 사용해요.

  • d1-import-tutorial이라는 데이터베이스.
  • TargetD1Table이라는 테이블.
  • TargetD1Table 안의 id, text, date_added라는 세 개의 컬럼.

테이블을 만들려면 다음 단계를 따르세요.

  1. Cloudflare 대시보드에서 D1 페이지로 이동하세요. D1 SQL database ↗로 이동

  2. Create database를 선택하세요.

  3. 데이터베이스 이름을 지정하세요. 이 튜토리얼에서는 D1 데이터베이스 이름을 d1-import-tutorial로 지정하세요.

  4. (선택) 위치 힌트(location hint)를 제공하세요. 위치 힌트는 데이터베이스의 원하는 지리적 위치를 나타내기 위해 제공할 수 있는 선택적 매개변수예요. 자세한 내용은 위치 힌트 제공을 참고하세요.

  5. Create를 선택하세요.

  6. Console로 이동한 다음 다음 SQL 스니펫을 붙여 넣으세요. 이 코드는 TargetD1Table이라는 테이블을 만들어요.

    DROP TABLE IF EXISTS TargetD1Table;
CREATE TABLE IF NOT EXISTS TargetD1Table (id INTEGER PRIMARY KEY, text TEXT, date_added TEXT);

    또는 Wrangler CLI를 사용해도 돼요.

    # Create a D1 database
npx wrangler d1 create d1-import-tutorial

# Create a D1 table
npx wrangler d1 execute d1-import-tutorial --command="DROP TABLE IF EXISTS TargetD1Table; CREATE TABLE IF NOT EXISTS TargetD1Table (id INTEGER PRIMARY KEY, text TEXT, date_added TEXT);" --remote

3. index.js 파일 만들기

  1. 새 디렉터리를 만들고 새 Node.js 프로젝트를 초기화하세요.

    mkdir d1-import-tutorial
cd d1-import-tutorial
npm init -y

  2. 이 저장소에 index.js라는 새 파일을 만드세요. 이 파일에는 REST API를 사용해 데이터를 D1 데이터베이스로 가져오는 코드가 담겨요.

  3. index.js 파일에 다음 변수를 정의하세요.

    • TARGET_TABLE: 대상 테이블 이름
    • ACCOUNT_ID: 계정 ID. Workers & Pages의 Account Details를 참고하세요.
    • DATABASE_ID: D1 데이터베이스 ID. 데이터베이스로 이동하면 데이터베이스 ID를 볼 수 있어요.
    • D1_API_KEY: 1단계에서 생성한 D1 API 토큰

    주의

    프로덕션에서는 민감한 정보를 저장할 때 환경 변수를 사용해야 해요.

    const TARGET_TABLE = " "; // for the tutorial, `TargetD1Table`
const ACCOUNT_ID = " ";
const DATABASE_ID = " ";
const D1_API_KEY = " ";
const D1_URL = `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/d1/database/${DATABASE_ID}/import`;
const filename = crypto.randomUUID(); // create a random filename
const uploadSize = 500;
const headers = {
	"Content-Type": "application/json",
	Authorization: *** ${D1_API_KEY}`,
};

4. 예제 데이터 생성하기 (선택)

실제로는 D1 데이터베이스로 가져오려는 데이터가 이미 있을 수 있어요.

이 튜토리얼에서는 가져오기 과정을 보여 주기 위해 예제 데이터를 생성해요.

  1. @faker-js/faker 모듈을 설치하세요.

    npm i @faker-js/faker

    yarn add @faker-js/faker

    pnpm add @faker-js/faker

    bun add @faker-js/faker

  2. index.js 파일의 시작 부분에 다음 코드를 추가하세요. 이 코드는 2500(uploadSize)개의 배열 요소를 가진 data라는 배열을 만드는데, 각 배열 요소는 id, text, date_added를 가진 객체를 담고 있어요. 각 배열 요소는 테이블 행에 해당해요.

    import crypto from "crypto";
import { faker } from "@faker-js/faker";

// Generate Fake data
const data = Array.from({ length: uploadSize }, () => ({
	id: Math.floor(Math.random() * 1000000),
	text: faker.lorem.paragraph(),
	date_added: new Date().toISOString().slice(0, 19).replace("T", " "),
}));

5. SQL 명령 생성하기

  1. 데이터를 대상 테이블에 삽입할 SQL 명령을 생성하는 함수를 만드세요. 이 함수는 이전 단계에서 생성한 data 배열을 사용해요.

    function makeSqlInsert(data, tableName, skipCols = []) {
	const columns = Object.keys(data[0]).join(",");
	const values = data
		.map((row) => {
			return (
				"(" +
				Object.values(row)
					.map((val) => {
						if (skipCols.includes(val) || val === null || val === "") {
							return "NULL";
						}
						return `'${String(val).replace(/'/g, "").replace(/"/g, "'")}'`;
					})
					.join(",") +
				")"
			);
		})
		.join(",");

	return `INSERT INTO ${tableName} (${columns}) VALUES ${values};`;
}

6. D1로 데이터 가져오기

가져오기 과정은 네 단계로 이뤄져요.

  1. Init upload: 이 단계는 업로드 과정을 초기화해요. SQL 명령의 해시를 D1 API에 보내고 업로드 URL을 받아요.

  2. Upload to R2: 이 단계는 SQL 명령을 업로드 URL에 업로드해요.

  3. Start ingestion: 이 단계는 수집(ingestion) 과정을 시작해요.

  4. Polling: 이 단계는 가져오기 과정이 끝날 때까지 폴링해요.

  5. 가져오기 과정의 네 단계를 실행하는 uploadToD1이라는 함수를 만드세요.

    async function uploadToD1() {
	// 1. Init upload
	const hashStr = crypto.createHash("md5").update(sqlInsert).digest("hex");

	try {
		const initResponse = await fetch(D1_URL, {
			method: "POST",
			headers,
			body: JSON.stringify({
				action: "init",
				etag: hashStr,
			}),
		});

		const uploadData = await initResponse.json();
		const uploadUrl = uploadData.result.upload_url;
		const filename = uploadData.result.filename;

		// 2. Upload to R2
		const r2Response = await fetch(uploadUrl, {
			method: "PUT",
			body: sqlInsert,
		});

		const r2Etag = r2Response.headers.get("ETag").replace(/"/g, "");

		// Verify etag
		if (r2Etag !== hashStr) {
			throw new Error("ETag mismatch");
		}

		// 3. Start ingestion
		const ingestResponse = await fetch(D1_URL, {
			method: "POST",
			headers,
			body: JSON.stringify({
				action: "ingest",
				etag: hashStr,
				filename,
			}),
		});

		const ingestData = await ingestResponse.json();
		console.log("Ingestion Response:", ingestData);

		// 4. Polling
		await pollImport(ingestData.result.at_bookmark);

		return "Import completed successfully";
	} catch (e) {
		console.error("Error:", e);
		return "Import failed";
	}
}

    위 코드에서:

    • SQL 명령의 md5 해시가 생성돼요.
    • initResponse는 업로드 과정을 초기화하고 업로드 URL을 받아요.
    • r2Response는 SQL 명령을 업로드 URL에 업로드해요.
    • 수집을 시작하기 전에 ETag가 검증돼요.
    • ingestResponse는 수집 과정을 시작해요.
    • pollImport는 가져오기 과정이 끝날 때까지 폴링해요.

  6. index.js 파일에 pollImport 함수를 추가하세요.

    async function pollImport(bookmark) {
	const payload = {
		action: "poll",
		current_bookmark: bookmark,
	};

	while (true) {
		const pollResponse = await fetch(D1_URL, {
			method: "POST",
			headers,
			body: JSON.stringify(payload),
		});

		const result = await pollResponse.json();
		console.log("Poll Response:", result.result);

		const { success, error } = result.result;

		if (
			success ||
			(!success && error === "Not currently importing anything.")
		) {
			break;
		}

		await new Promise((resolve) => setTimeout(resolve, 1000));
	}
}

    위 코드는 다음과 같은 일을 해요.

    • D1 API에 poll 액션을 보내요.
    • 가져오기 과정이 끝날 때까지 폴링해요.

  7. 마지막으로 가져오기 과정을 실행할 runImport 함수를 index.js 파일에 추가하세요.

    async function runImport() {
	const result = await uploadToD1();
	console.log(result);
}

runImport();

7. 최종 코드 작성하기

이전 단계에서 D1로 데이터를 가져오는 데 관련된 여러 과정을 실행하는 함수들을 만들었어요. 최종 코드는 이 함수들을 실행해서 예제 데이터를 대상 D1 테이블로 가져와요.

  1. 코드 맨 위에 변수를 정의하고, index.js 파일의 최종 코드를 아래와 같이 복사하세요.

    import crypto from "crypto";
import { faker } from "@faker-js/faker";

const TARGET_TABLE = "";
const ACCOUNT_ID = "";
const DATABASE_ID = "";
const D1_API_KEY = "";
const D1_URL = `https://api.cloudflare.com/client/v4/accounts/${ACCOUNT_ID}/d1/database/${DATABASE_ID}/import`;
const uploadSize = 500;
const headers = {
	"Content-Type": "application/json",
	Authorization: *** ${D1_API_KEY}`,
};

// Generate Fake data
const data = Array.from({ length: uploadSize }, () => ({
	id: Math.floor(Math.random() * 1000000),
	text: faker.lorem.paragraph(),
	date_added: new Date().toISOString().slice(0, 19).replace("T", " "),
}));

// Make SQL insert statements
function makeSqlInsert(data, tableName, skipCols = []) {
	const columns = Object.keys(data[0]).join(",");
	const values = data
		.map((row) => {
			return (
				"(" +
				Object.values(row)
					.map((val) => {
						if (skipCols.includes(val) || val === null || val === "") {
							return "NULL";
						}
						return `'${String(val).replace(/'/g, "").replace(/"/g, "'")}'`;
					})
					.join(",") +
				")"
			);
		})
		.join(",");

	return `INSERT INTO ${tableName} (${columns}) VALUES ${values};`;
}

const sqlInsert = makeSqlInsert(data, TARGET_TABLE);

async function pollImport(bookmark) {
	const payload = {
		action: "poll",
		current_bookmark: bookmark,
	};

	while (true) {
		const pollResponse = await fetch(D1_URL, {
			method: "POST",
			headers,
			body: JSON.stringify(payload),
		});

		const result = await pollResponse.json();
		console.log("Poll Response:", result.result);

		const { success, error } = result.result;

		if (
			success ||
			(!success && error === "Not currently importing anything.")
		) {
			break;
		}

		await new Promise((resolve) => setTimeout(resolve, 1000));
	}
}

// Upload to D1
async function uploadToD1() {
	// 1. Init upload
	const hashStr = crypto.createHash("md5").update(sqlInsert).digest("hex");

	try {
		const initResponse = await fetch(D1_URL, {
			method: "POST",
			headers,
			body: JSON.stringify({
				action: "init",
				etag: hashStr,
			}),
		});

		const uploadData = await initResponse.json();
		const uploadUrl = uploadData.result.upload_url;
		const filename = uploadData.result.filename;

		// 2. Upload to R2
		const r2Response = await fetch(uploadUrl, {
			method: "PUT",
			body: sqlInsert,
		});

		const r2Etag = r2Response.headers.get("ETag").replace(/"/g, "");

		// Verify etag
		if (r2Etag !== hashStr) {
			throw new Error("ETag mismatch");
		}

		// 3. Start ingestion
		const ingestResponse = await fetch(D1_URL, {
			method: "POST",
			headers,
			body: JSON.stringify({
				action: "ingest",
				etag: hashStr,
				filename,
			}),
		});

		const ingestData = await ingestResponse.json();
		console.log("Ingestion Response:", ingestData);

		// 4. Polling
		await pollImport(ingestData.result.at_bookmark);

		return "Import completed successfully";
	} catch (e) {
		console.error("Error:", e);
		return "Import failed";
	}
}

async function runImport() {
	const result = await uploadToD1();
	console.log(result);
}

runImport();

8. 코드 실행하기

  1. 코드를 실행하세요.

    node index.js

이제 대상 D1 테이블에 예제 데이터가 채워진 것을 볼 수 있을 거예요.

참고

statement too long 오류가 발생하면 SQL 명령을 더 작은 조각으로 나눠 배치로 업로드해야 해요. 이 오류에 대해 더 자세히 배우려면 D1 문서를 참고하세요.

요약 (Summary)

이 튜토리얼을 완료하면서 여러분은 다음과 같은 일을 했어요.

  1. API 토큰을 만들었어요.
  2. 대상 데이터베이스와 테이블을 만들었어요.
  3. 예제 데이터를 생성했어요.
  4. 예제 데이터에 대한 SQL 명령을 만들었어요.
  5. REST API를 사용해 예제 데이터를 D1 대상 테이블로 가져왔어요.

더 알아보기 (Learn more)