Skip to content

Runnable recipes / 场景示例

Documentation home · 中文手册 · API reference

These recipes use a disposable, initially empty table with a cf column family. Each recipe is independent. Create/open a different table for each recipe using the ownership pattern in the complete example, then run the block with client, table, and table_name in scope. Remove only your own table afterward. Do not run these writes against an important table.

The Python blocks marked runnable include assertions and were executed directly from this Markdown during local HBase validation. This documentation-only release does not add Python test files; the published site workflow builds documentation, not a live HBase cluster. See verification scope. Passing these examples does not validate every cluster topology or failure mode.

1. Store a UTF-8 profile and JSON / 中文与 JSON

Use explicit encoding at the application boundary. Missing rows, missing columns, and empty values are different cases.

import json

from hbasedriver.operations.get import Get
from hbasedriver.operations.put import Put

profile = {"name": "张三", "roles": ["reader"]}
table.put(
    Put(b"user:001")
    .add_column(b"cf", b"name", profile["name"].encode("utf-8"))
    .add_column(b"cf", b"profile", json.dumps(profile, ensure_ascii=False).encode("utf-8"))
    .add_column(b"cf", b"note", b"")
)
row = table.get(Get(b"user:001"))
assert row is not None
assert row.get(b"cf", b"name").decode("utf-8") == "张三"
assert json.loads(row.get(b"cf", b"profile").decode("utf-8")) == profile
assert row.get(b"cf", b"note") == b""
assert row.get(b"cf", b"missing") is None
assert table.get(Get(b"missing-row")) is None

Row.kv is nested: {family: {qualifier: value}}. It is not HappyBase's flat {b"cf:q": value} mapping.

2. Filter active users / 组合过滤

Combine a row-key prefix and a column-value predicate. filter_if_missing=True excludes rows without the tested column. FilterList.Operator is nested inside FilterList; there is no public FilterListOperator import.

from hbasedriver.filter import (
    BinaryComparator, CompareOperator, FilterList, PrefixFilter, QualifierFilter,
    SingleColumnValueFilter,
)
from hbasedriver.operations.put import Put
from hbasedriver.operations.scan import Scan

for key, state in [(b"user:001", b"active"), (b"user:002", b"inactive")]:
    table.put(Put(key).add_column(b"cf", b"state", state))
table.put(Put(b"user:003").add_column(b"cf", b"name", b"No state"))

filters = FilterList(
    operator=FilterList.Operator.MUST_PASS_ALL,
    filters=[
        PrefixFilter(b"user:"),
        SingleColumnValueFilter(
            b"cf", b"state", CompareOperator.EQUAL, BinaryComparator(b"active"),
            filter_if_missing=True, latest_version_only=True,
        ),
    ],
)
scan = (
    Scan(start_row=b"user:")
    .with_end_row(b"user;", inclusive=False)
    .add_family(b"cf")
    .set_filter(filters)
)
with table.scan(scan) as scanner:
    rows = list(scanner)
assert [row.rowkey for row in rows] == [b"user:001"]

# Current Scan.add_column() cannot serialize a bytes qualifier.
# A family plus a qualifier filter selects the state cells instead.
projection = (
    Scan(start_row=b"user:")
    .with_end_row(b"user;", inclusive=False)
    .add_family(b"cf")
    .set_filter(QualifierFilter(CompareOperator.EQUAL, BinaryComparator(b"state")))
)
with table.scan(projection) as scanner:
    selected = scanner.next_batch(10)
assert [row.rowkey for row in selected] == [b"user:001", b"user:002"]
assert all(set(row.kv[b"cf"]) == {b"state"} for row in selected)

The predicate applies on the server during Scan. Do not copy it to Get.set_filter(): that filter is not sent by the current Get path. More selection patterns: advanced usage.

3. Paginate without repeating the last row / 游标分页

Use an exclusive start after the last returned key, and preserve the same end bound and filter on every page. Unlike some HBase APIs, this driver's constructor includes both endpoints by default.

from hbasedriver.filter import PrefixFilter
from hbasedriver.operations.put import Put
from hbasedriver.operations.scan import Scan

expected = [f"page:{number:03d}".encode("ascii") for number in range(5)]
for key in expected:
    table.put(Put(key).add_column(b"cf", b"value", key))

resume_key = None
seen = []
while True:
    scan = (
        Scan(start_row=b"page:")
        .with_end_row(b"page;", inclusive=False)
        .add_family(b"cf")
        .set_filter(PrefixFilter(b"page:"))
    )
    if resume_key is not None:
        scan.with_start_row(resume_key, inclusive=False)
    rows, resume_key = table.scan_page(scan, page_size=2)
    if not rows:
        break
    seen.extend(row.rowkey for row in rows)

assert seen == expected
assert len(seen) == len(set(seen))

A short final page can still return a resume key; terminate on an empty page. This is not a consistent snapshot across concurrent writes, nor a promise of reverse/cross-region pagination behavior under topology changes. Keep cursors as bytes internally; encode them explicitly (for example base64) when transporting them as JSON.

4. Batch input and keyed results / 批量读写

BatchGet.add_column() does not return the batch. Read results are keyed by row key, not a list of Row objects. Duplicate requested keys collapse in the result dictionary.

from hbasedriver.operations.batch import BatchGet, BatchPut
from hbasedriver.operations.put import Put

writes = BatchPut()
for key, value in [(b"batch:1", b"A"), (b"batch:2", b"B")]:
    writes.add_put(Put(key).add_column(b"cf", b"value", value))
statuses = table.batch_put(writes)
assert statuses == [True, True]

reads = BatchGet([b"batch:1", b"batch:2", b"batch:missing"])
reads.add_column(b"cf", b"value")
results = table.batch_get(reads)
assert results[b"batch:1"].get(b"cf", b"value") == b"A"
assert results[b"batch:2"].get(b"cf", b"value") == b"B"
assert results[b"batch:missing"] is None

These are convenience APIs: the current implementation issues per-row operations, not one combined RPC or a cross-row transaction. A failure can occur after earlier writes have succeeded.

5. Counters and create-if-absent / 计数与条件写入

Let HBase initialize an absent counter. Never initialize it with text b"0": HBase counters are signed 8-byte big-endian integers. Conditional value=None means the cell must be absent; b"" means an existing empty value.

from hbasedriver.operations.get import Get
from hbasedriver.operations.increment import CheckAndPut, Increment
from hbasedriver.operations.put import Put

first_count = table.increment(Increment(b"counter:1").add_column(b"cf", b"count", 1))
second_count = table.increment(Increment(b"counter:1").add_column(b"cf", b"count", 2))
assert first_count == 1
assert second_count == 3
counter = table.get(Get(b"counter:1"))
assert counter is not None
assert int.from_bytes(counter.get(b"cf", b"count"), "big", signed=True) == 3

claim = CheckAndPut(b"job:1")
claim.set_check(b"cf", b"state", None)
claim.set_put(Put(b"job:1").add_column(b"cf", b"state", b"claimed"))
first_claim = table.check_and_put(claim)
second_claim = table.check_and_put(claim)
assert first_claim is True
assert second_claim is False

This illustrates atomic comparison, not a complete distributed lock (no lease or fencing token). Retrying increments after an ambiguous timeout can count twice. See failure semantics.

6. Append and same-row mutations / 追加与单行原子修改

Append concatenates bytes. RowMutations groups operations on the same row. Use different qualifiers for independent changes.

from hbasedriver.operations.append import Append
from hbasedriver.operations.delete import Delete
from hbasedriver.operations.get import Get
from hbasedriver.operations.put import Put
from hbasedriver.operations.row_mutations import RowMutations

key = b"event:1"
table.put(
    Put(key)
    .add_column(b"cf", b"tags", b"created")
    .add_column(b"cf", b"temporary", b"yes")
)
updated = table.append(Append(key).add_column(b"cf", b"tags", b",processed"))
assert updated is not None
assert updated.get(b"cf", b"tags") == b"created,processed"

changes = RowMutations(key)
changes.add(Put(key).add_column(b"cf", b"state", b"done"))
changes.add(Delete(key).add_column(b"cf", b"temporary"))
applied = table.mutate_row(changes)
assert applied is True
row = table.get(Get(key))
assert row is not None
assert row.get(b"cf", b"state") == b"done"
assert row.get(b"cf", b"temporary") is None

This does not create atomicity across rows. A repeated Append repeats its suffix; design retry/reconciliation explicitly.

7. Flush buffered writes / 缓冲写入

Construct parameters first, then assign attributes. The flush interval is in milliseconds, and zero disables the background flush thread. The caller owns table; closing the mutator does not close it.

from hbasedriver.client.buffered_mutator import BufferedMutator, BufferedMutatorParams
from hbasedriver.operations.get import Get
from hbasedriver.operations.put import Put

params = BufferedMutatorParams()
params.write_buffer_size = 1024 * 1024
params.write_buffer_periodic_flush = 0
with BufferedMutator(table, params) as mutator:
    for number in range(3):
        mutator.mutate(
            Put(f"buffer:{number}".encode("ascii"))
            .add_column(b"cf", b"value", b"saved")
        )
    assert mutator.get_current_buffer_size_mutations() == 3
    mutator.flush()
    assert mutator.get_current_buffer_size_mutations() == 0

for number in range(3):
    row = table.get(Get(f"buffer:{number}".encode("ascii")))
    assert row is not None
    assert row.get(b"cf", b"value") == b"saved"

Buffering defers execution; it does not currently combine mutations into a single RPC. It is not durable: a failed flush can leave a partially written batch with no retained pending queue. Preserve recoverable input outside the mutator.

8. Existence checks and deletion / 存在性与删除

Use fresh Get objects for existence checks because the method changes their existence-only flag. This recipe writes one version of each cell, so deleting one version makes that cell disappear.

from hbasedriver.operations.delete import Delete
from hbasedriver.operations.get import Get
from hbasedriver.operations.put import Put

key = b"delete:1"
table.put(
    Put(key)
    .add_column(b"cf", b"keep", b"A")
    .add_column(b"cf", b"remove", b"B")
)
assert table.exists(Get(key)) is True
assert table.exists_all([Get(key), Get(b"absent")]) == {key: True, b"absent": False}

table.delete(Delete(key).add_column(b"cf", b"remove"))
row = table.get(Get(key))
assert row is not None
assert row.get(b"cf", b"keep") == b"A"
assert row.get(b"cf", b"remove") is None

table.delete(Delete(key))
assert table.get(Get(key)) is None

For versioned data, Delete.add_column() deletes one version; Delete.add_columns(cf, qualifier, timestamp) requires an explicit timestamp and targets versions through that timestamp. Test retention and deletion against the real table schema before migration.