外观
写入数据
选择如何建表、追加记录或修改已有数据。
适用: Embedded ✅、Enterprise ✅
写入(ingestion)就是把数据带进一张 LanceDB 表。先确定 schema 并建表,然后根据新记录与已有数据的关系,选择对应的写操作。
选择写操作
| 任务 | 操作 | 效果 |
|---|---|---|
| 创建表 | create_table / createTable | 用数据或显式 schema 建一张表。 |
| 追加记录 | add | 直接加行,不与已有键做匹配。 |
| 按 key 更新插入 | merge_insert / mergeInsert | 自行选择命中的行是更新、新键是插入。 |
| 更新选中的行 | update | 修改被谓词选中的行里的值。 |
| 删除选中的行 | delete | 从表的当前内容里移除命中的行。 |
同一批输入追加两次会产生重复记录。如果新数据应当更新已有记录,请用带稳定 key 的 merge insert。
连接并准备数据
连接 Embedded 或 Enterprise 用连接指南。本节里的 SDK 写操作在两者上都能用。Enterprise 另外支持 SQL 写语句;而 SDK 的过滤表达式不需要另开一条 SQL 连接。
在开始大批量导入之前,先定好列类型、必填字段与向量维度。各 SDK 支持的输入形式不同,包括行对象与 Arrow 批次这些选择见建表指南。
跑一遍完整的写入流程
这个示例从 Alice 和 Bob 开始,追加 Charlie,更新 Bob 的名字与登录次数,再合并一批包含 Charlie 和 Dana 的数据。最后把 Dana 那行查出来看一眼,然后删掉。
完整示例:创建、追加、更新、Upsert 与删除
每个程序都会建一个临时本地数据库,依次执行这五种操作,打印结果,跑完清理掉临时数据。
- Python: 安装
lancedb,把代码存成ingestion.py,执行python ingestion.py。 - TypeScript: 安装
@lancedb/lancedb,把代码存成ingestion.mts,用 Node.js 22.18 或更高版本执行node ingestion.mts。 - Rust: 下载完整的 Rust 工程,解压后执行
cargo run --locked。压缩包里有src/main.rs、Cargo manifest 以及示例依赖的 lockfile。代码见下。
python
from tempfile import TemporaryDirectory
import lancedb
with TemporaryDirectory(prefix="lancedb-ingestion-") as uri:
db = lancedb.connect(uri)
table = db.create_table(
"users",
[
{"id": 1, "name": "Alice", "login_count": 10},
{"id": 2, "name": "Bob", "login_count": 20},
],
)
table.add([{"id": 3, "name": "Charlie", "login_count": 5}])
table.update(where="id = 2", values={"name": "Bobby"})
table.update(where="id = 2", values_sql={"login_count": "login_count + 1"})
(
table.merge_insert("id")
.when_matched_update_all()
.when_not_matched_insert_all()
.execute(
[
{"id": 3, "name": "Charlie", "login_count": 6},
{"id": 4, "name": "Dana", "login_count": 1},
]
)
)
print("Rows selected for deletion:", table.search().where("id = 4").to_list())
table.delete("id = 4")
rows = sorted(table.search().to_list(), key=lambda row: row["id"])
print(rows)typescript
const lancedb = await import("@lancedb/lancedb");
const { mkdtemp, rm } = await import("node:fs/promises");
const { tmpdir } = await import("node:os");
const { join } = await import("node:path");
const uri = await mkdtemp(join(tmpdir(), "lancedb-ingestion-"));
try {
const db = await lancedb.connect(uri);
const table = await db.createTable("users", [
{ id: 1, name: "Alice", login_count: 10 },
{ id: 2, name: "Bob", login_count: 20 },
]);
await table.add([{ id: 3, name: "Charlie", login_count: 5 }]);
await table.update({ where: "id = 2", values: { name: "Bobby" } });
await table.update({
where: "id = 2",
valuesSql: { login_count: "login_count + 1" },
});
await table
.mergeInsert("id")
.whenMatchedUpdateAll()
.whenNotMatchedInsertAll()
.execute([
{ id: 3, name: "Charlie", login_count: 6 },
{ id: 4, name: "Dana", login_count: 1 },
]);
console.log(
"Rows selected for deletion:",
await table.query().where("id = 4").toArray(),
);
await table.delete("id = 4");
const rows = (await table.query().toArray()).sort((a, b) => a.id - b.id);
console.log(rows);
} finally {
await rm(uri, { recursive: true, force: true });
}rust
use std::sync::Arc;
use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, RecordBatchReader, StringArray};
use arrow_schema::{DataType, Field, Schema};
use futures::TryStreamExt;
use lancedb::query::{ExecutableQuery, QueryBase};
fn make_users_reader(rows: &[(i64, &str, i64)]) -> Box<dyn RecordBatchReader + Send> {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("name", DataType::Utf8, false),
Field::new("login_count", DataType::Int64, false),
]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.0))),
Arc::new(StringArray::from_iter_values(rows.iter().map(|r| r.1))),
Arc::new(Int64Array::from_iter_values(rows.iter().map(|r| r.2))),
],
)
.unwrap();
Box::new(RecordBatchIterator::new(
vec![Ok(batch)].into_iter(),
schema,
))
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let directory = tempfile::tempdir()?;
let db = lancedb::connect(directory.path().to_str().unwrap())
.execute()
.await?;
let table = db
.create_table(
"users",
make_users_reader(&[(1, "Alice", 10), (2, "Bob", 20)]),
)
.execute()
.await?;
table
.add(make_users_reader(&[(3, "Charlie", 5)]))
.execute()
.await?;
table
.update()
.only_if("id = 2")
.column("name", "'Bobby'")
.execute()
.await?;
table
.update()
.only_if("id = 2")
.column("login_count", "login_count + 1")
.execute()
.await?;
let mut merge = table.merge_insert(&["id"]);
merge
.when_matched_update_all(None)
.when_not_matched_insert_all();
merge
.execute(make_users_reader(&[(3, "Charlie", 6), (4, "Dana", 1)]))
.await?;
let selected: Vec<RecordBatch> = table
.query()
.only_if("id = 4")
.execute()
.await?
.try_collect()
.await?;
println!("Rows selected for deletion: {selected:?}");
table.delete("id = 4").await?;
let rows: Vec<RecordBatch> = table.query().execute().await?.try_collect().await?;
println!("Remaining rows: {rows:?}");
assert_eq!(table.count_rows(None).await?, 3);
assert_eq!(
table
.count_rows(Some(
"id = 2 AND name = 'Bobby' AND login_count = 21".into()
))
.await?,
1
);
assert_eq!(
table
.count_rows(Some("id = 3 AND login_count = 6".into()))
.await?,
1
);
Ok(())
}这些操作跑完之后,Alice 有 10 次登录,Bobby 有 21 次,Charlie 有 6 次。Dana 被选中删除,最终结果里没有它。
校验结果
一次写入完成后,检查受影响的键与字段,并核对行数。SDK 的返回值随操作和语言而异;如果你要依赖返回的计数或版本元数据,请查对应的 API 参考。
持有旧快照的读取方,可能不会立刻看到另一个连接句柄的写入,详见一致性。一张被 checkout 到历史版本的表,必须先回到最新版本才能被修改。
加载更大的数据集
当数据集大到放不进内存时,改用批次或文件型输入。插入指南里讲了输入格式、进度上报与写入并行度。也要注意重试语义:重复一次追加,和重试一次基于 key 的 upsert,并不是一回事。