MaxStorage

MaxCompute MaxStorage 是 MaxCompute 提供的高吞吐数据读写接口。相比于基于 Tunnel 的数据通道， MaxStorage 提供了更细粒度的会话管理、支持 Arrow 格式和 Blob 格式读写、支持增量读取以及表预览等功能，适用于大规模并行数据读写场景。

备注

MaxStorage 需要服务端支持，请确保 MaxCompute 集群已开启相关功能。

基本概念

MaxStorage 的核心概念包括：

会话（Session）：读写操作的事务上下文。读会话管理数据分片，写会话保证数据原子性。
分片（Split）：读会话将数据按大小、并行度等方式分为多个分片，每个分片可独立读取，支持并行处理。
流（Stream）：写会话中的数据上传通道。一个写会话可创建多个流以支持并行写入。
压缩（Compression）：读路径支持 UNCOMPRESSED（默认）、LZ4 和 ZSTD 压缩算法，写路径通过 Arrow IPC 内置压缩实现，减少网络传输数据量。
路由令牌（Route Token）：服务端返回的路由标识，用于会话亲和性，确保后续请求路由到同一节点。
Exactly-Once 模式：写流支持精确一次语义，通过 access_token 和 row_offset 实现幂等写入。
API 版本：api_version 参数（默认 "2"）选择 URL 路径 api/storage/v2 或 api/storage/v3，v3 特性通过 _supports_v3() 门控。

客户端初始化

使用 MaxStorage 需要创建 MaxStorageClient 实例，传入 ODPS 入口对象。

from odps import ODPS
from odps.maxstorage import MaxStorageClient

# 初始化 ODPS 入口对象
odps = ODPS(
    access_id="your_access_id",
    secret_access_key="your_secret_access_key",
    project="your_project",
    endpoint="your_endpoint",
)

# 创建 MaxStorage 客户端，默认 api_version="2"
client = MaxStorageClient(odps)

# 使用 API v3（开启 WriteMode、嵌套 Blob 等高级特性）
client_v3 = MaxStorageClient(odps, api_version="3")

也可以通过 tunnel_endpoint 参数手动指定 Tunnel 端点，绕过自动发现：

client = MaxStorageClient(odps, tunnel_endpoint="http://your-tunnel-endpoint")

还可以通过 quota_name 参数指定资源配额：

client = MaxStorageClient(odps, quota_name="your_quota")

读取数据

完整读取流程

使用 MaxStorage 读取数据需要以下步骤：

创建读会话
按分片读取数据
关闭读会话（会话会自动过期，无需手动关闭）

创建读会话

通过 create_table_read_session() 方法创建读会话。读会话确定了数据的分片方式、返回的列和分区等。创建后会轮询直到会话状态变为 NORMAL。

from odps.maxstorage import MaxStorageClient, SplitOptions, SplitMode

client = MaxStorageClient(odps)

# 创建读会话，使用默认分片选项（按大小分片）
read_session = client.create_table_read_session("your_table")
print(f"Session ID: {read_session.id}")
print(f"数据分片数: {len(read_session.splits)}")
print(f"总记录数: {read_session.record_count}")
print(f"过期时间: {read_session.expiration_time}")

备注

read_session.record_count 取自服务端返回的 RecordCount。若服务端未返回该字段，该属性为 -1，表示总行数未知，而不是 0。

创建读会话时可以指定需要读取的列、分区和分片选项：

# 仅读取指定列和分区
read_session = client.create_table_read_session(
    "your_table",
    columns=["id", "name", "value"],
    partitions=["pt=20230101"],
)

# 按行偏移分片，每个分片 100 万行
split_opts = SplitOptions(
    split_mode=SplitMode.ROW_OFFSET,
    split_number=1000000,
)
read_session = client.create_table_read_session(
    "your_table",
    split_options=split_opts,
)

备注

read_session.splits 返回的分片类型取决于 SplitMode： SplitMode.SIZE 下为 ROW_OFFSET` 下为 open_arrow_reader。

备注

对于 append2.0 / 事务表，服务端仅支持 SplitMode.ROW_OFFSET 分片模式。

使用 Arrow 格式读取数据

通过 open_arrow_reader() 方法可以直接读取为 Arrow RecordBatch：

import pyarrow as pa

read_session = client.create_table_read_session("your_table")

# 遍历所有分片读取数据
for split in read_session.splits:
    reader = read_session.open_arrow_reader(split)
    while True:
        batch = reader.read()
        if batch is None:
            break
        df = batch.to_pandas()
        # 处理 DataFrame
    reader.close()

读取时可以通过 max_batch_rows 控制每个批次的行数，通过 skip_row_num 跳过指定行数：

reader = read_session.open_arrow_reader(
    read_session.splits[0],
    max_batch_rows=1024,
    skip_row_num=100,
)

也可以通过 get_as_record_reader() 获取行形式的 Record 迭代器。返回的 ArrowRecordReader 除了迭代外，还暴露以下属性：

schema：返回 OdpsSchema，包含列名与类型信息。
count：返回已读取的记录数，迭代结束后等于该分片的总行数。

reader = read_session.open_arrow_reader(read_session.splits[0])
rr = reader.get_as_record_reader()
print(rr.schema.columns)  # 列信息
for record in rr:
    print(record[0], record[1])
print(rr.count)           # 已读取的记录数

扩展 Arrow IPC（Extended Arrow IPC）

服务端可能为读会话协商**扩展 Arrow IPC** 流格式：响应流由若干带标签的子流（substream）拼接而成，标签 1 为数据子流（其中的批次才是表数据行），标签 2 为指标（metrics）子流。可通过 extended_arrow_ipc_enabled 判断当前会话是否启用了该格式：

read_session = client.create_table_read_session("your_table")
print(read_session.extended_arrow_ipc_enabled)

会话启用该格式时，open_arrow_reader() 会自动切换到扩展读取器，调用方无需额外参数：

迭代与 read() 只返回数据子流中的批次，指标子流不会作为数据行出现。
指标子流携带的 schema metadata 会合并到 metrics_metadata；普通（非扩展）读取器该属性为 None。
多个指标子流中的耗时指标（server_processing_duration_us、rate_limit_duration_us）会累加，其余键保留最后一次观测到的值。指标在整个分片消费完毕后才是完整的。

reader = read_session.open_arrow_reader(read_session.splits[0])
for batch in reader:
    process(batch)
print(reader.metrics_metadata)   # 例如 {"server_processing_duration_us": "120"}
reader.close()

并行读取数据

MaxStorage 的分片机制天然支持并行读取。每个分片可以独立读取，适合多线程场景：

from concurrent.futures import ThreadPoolExecutor

read_session = client.create_table_read_session("your_table")

def read_split(split):
    reader = read_session.open_arrow_reader(split)
    batches = []
    while True:
        batch = reader.read()
        if batch is None:
            break
        batches.append(batch)
    reader.close()
    return pa.concat_batches(batches) if batches else None

# 使用线程池并行读取所有分片
with ThreadPoolExecutor(max_workers=len(read_session.splits)) as pool:
    futures = [pool.submit(read_split, s) for s in read_session.splits]
    results = [f.result() for f in futures]

增量读取

增量读取用于只读取自某个版本以来新增或变更的数据：

from odps.maxstorage import IncrementalReadOptions

incr_opts = IncrementalReadOptions(
    version="v1",
    from_=100,
    to=200,
)
read_session = client.create_table_read_session(
    "your_table",
    incremental_read_enabled=True,
    incremental_read_options=incr_opts,
)

IncrementalReadOptions.start_txn_id 用于请求精确一次（exactly-once）的增量读取：服务端从给定事务 ID 之后开始返回数据。读会话的 latest_txn_id 属性返回服务端报告的最新事务 ID（LatestTxnId，缺省为 None），可作为下次续读的 start_txn_id：

incr_opts = IncrementalReadOptions(version="Auto", start_txn_id="txn-1")
read_session = client.create_table_read_session(
    "your_table",
    incremental_read_enabled=True,
    incremental_read_options=incr_opts,
)
print(read_session.latest_txn_id)   # 服务端最新事务 ID，可用于续读

压缩读取

读路径压缩通过 compress_option 参数启用，默认为不压缩。支持 ZSTD 和 LZ4_FRAME。

备注

使用压缩前需安装对应的库：

ZSTD：pip install zstandard
LZ4：pip install lz4

from odps.tunnel import CompressOption

compress_option = CompressOption(
    CompressOption.CompressAlgorithm.ODPS_ZSTD,
)
reader = read_session.open_arrow_reader(
    read_session.splits[0],
    compress_option=compress_option,
)

也可以使用 compress_algo 简写：

reader = read_session.open_arrow_reader(
    read_session.splits[0],
    compress_algo="zstd",
)

读取 Instance 结果

通过 create_instance_read_session() 可以读取 SQL Instance 的结果：

instance = odps.execute_sql("SELECT * FROM your_table LIMIT 100")
instance_session = client.create_instance_read_session(instance)

reader = instance_session.open_arrow_reader(offset=0, count=100)
while True:
    batch = reader.read()
    if batch is None:
        break
    print(batch.to_pandas())
reader.close()

表预览

通过 preview_table() 可以快速预览表的前若干行，无需创建读会话：

reader = client.preview_table("your_table", limit=10)
while True:
    batch = reader.read()
    if batch is None:
        break
    print(batch.to_pandas())
reader.close()

写入数据

完整写入流程

使用 MaxStorage 写入数据需要以下步骤：

创建写会话
创建写流
写入数据（批量或记录形式）
提交会话

创建写会话

通过 create_table_write_session() 方法创建写会话：

from odps.maxstorage import MaxStorageClient, WriteMode

client = MaxStorageClient(odps)

# 创建批处理写会话（默认）
write_session = client.create_table_write_session("your_table")

# 创建流式写会话
write_session = client.create_table_write_session(
    "your_table",
    write_mode=WriteMode.STREAMING,
)

# 创建兼容模式写会话
write_session = client.create_table_write_session(
    "your_table",
    write_mode=WriteMode.BATCH_COMPATIBLE,
)

# 写入指定分区
write_session = client.create_table_write_session(
    "your_table",
    partition_spec="pt=20230101",
)

批量写入 Arrow RecordBatch

通过 open_arrow_writer() 创建写流，再调用 write_batch 写入：

import pyarrow as pa

write_session = client.create_table_write_session("your_table")
writer = write_session.open_arrow_writer(stream_id="0")

schema = pa.schema([
    ("id", pa.int64()),
    ("name", pa.string()),
])
batch = pa.RecordBatch.from_arrays(
    [pa.array([1, 2, 3], type=pa.int64()),
     pa.array(["a", "b", "c"], type=pa.string())],
    schema=schema,
)
writer.write_batch(batch)
writer.close()
write_session.commit()

异步刷新

通过 flush_async() 可以异步刷新缓冲区，配合 max_pending_buffers 控制背压：

writer = write_session.open_arrow_writer(stream_id="0", max_pending_buffers=4)
for batch in batches:
    writer.write_batch(batch)
    writer.flush_async()
writer.close()
write_session.commit()

记录形式写入

通过 get_as_record_writer() 可以使用行形式的 Record 接口写入，无需手动构造 Arrow 批次。该方法可用于任意 :class:`TableArrowWriter`（包括普通表和 BLOB 表）。

**普通表写入**（无 BLOB 列）：

from odps.models import Record

write_session = client.create_table_write_session("your_table")
writer = write_session.open_arrow_writer(stream_id="0")
record_writer = writer.get_as_record_writer()

record_writer.write(Record(columns=["id", "name"], values=[1, "alice"]))
record_writer.write(Record(columns=["id", "name"], values=[2, "bob"]))
record_writer.close()
write_session.commit()

**Delta 表写入**（含 __operation 列，支持 UPSERT/DELETE）：

from odps.models import Record

write_session = client.create_table_write_session("delta_table")
writer = write_session.open_arrow_writer(stream_id="0")
record_writer = writer.get_as_record_writer()  # 返回 DeltaTableRecordWriter

record_writer.write(Record(columns=["id", "name"], values=[1, "alice"]))  # UPSERT
record_writer.write(Record(columns=["id", "name"], values=[2, "bob"]))    # UPSERT
record_writer.delete(Record(columns=["id", "name"], values=[2, None]))    # DELETE
record_writer.close()
write_session.commit()

备注

get_as_record_writer() 根据 schema 自动选择 writer 类型：若表包含 __operation 列，返回 DeltaTableRecordWriter，其 write() 方法标记 UPSERT、delete() 方法标记 DELETE，两种操作可在同一 writer 实例上自由交替使用。否则返回 AppendTableRecordWriter。 BLOB 表的 Record API 写入见自动上传。

压缩写入

写路径通过 Arrow IPC 内置压缩实现，通过 compress_option 参数启用：

from odps.tunnel import CompressOption

compress_option = CompressOption(
    CompressOption.CompressAlgorithm.ODPS_ZSTD,
)
writer = write_session.open_arrow_writer(
    stream_id="0",
    compress_option=compress_option,
)

提交与终止

写入完成后调用 commit() 提交数据。若需要丢弃数据，调用 abort()：

# 提交
write_session.commit()

# 终止（丢弃所有已上传数据）
write_session.abort()

备注

close() 不会自动提交：若未显式提交或终止，close() 会自动终止会话。

兼容模式写入（BatchCompatible）

WriteMode.BATCH_COMPATIBLE 使用 block_number + attempt_number 标识 Writer，并以强类型 BlockWriteResult 汇总提交，适合迁移依赖 Block 写入语义的任务。

兼容模式下不使用 open_arrow_writer，而是通过 open_block_writer() 创建 TableBlockWriter：

from odps.maxstorage import MaxStorageClient, WriteMode, BatchCompatibleOptions

client = MaxStorageClient(odps)
write_session = client.create_table_write_session(
    "your_table",
    write_mode=WriteMode.BATCH_COMPATIBLE,
    batch_compatible_options=BatchCompatibleOptions(
        enhance_write_check=True,
        max_field_size=8 * 1024 * 1024,
        dynamic_partition_limit=-1,
    ),
)

results = []
for block in range(block_count):
    writer = write_session.open_block_writer(block, 0)
    batch = pa.RecordBatch.from_arrays(
        [pa.array([1, 2, 3], type=pa.int64())],
        schema=writer.schema,
    )
    writer.write_batch(batch)
    results.append(writer.commit())

write_session.commit(block_results=results)

提交时可通过 commit 的 wait_flying_writers_timeout_seconds 参数指定服务端等待在途（in-flight）Writer 的超时时间（1..600 秒的整数）。若提交时仍有其他 Writer 在写入，服务端会在该时间窗口内等待其完成后才最终确定提交结果；不设置该参数时使用服务端默认行为，请求内容保持不变。该参数仅在兼容模式下生效，传入其他写入模式会抛出 ValueError：

write_session.commit(
    block_results=results,
    wait_flying_writers_timeout_seconds=120,
)

备注

设置该参数后，SDK 会自动将本次提交请求的客户端读超时提升为 max(options.read_timeout, wait_flying_writers_timeout_seconds + 30)，确保客户端不会在服务端仍处于合法等待窗口内时提前超时并重发提交请求。

备注

block_number 和 attempt_number 均从 0 开始。重试同一份 Block 数据时保持 block_number 不变并递增 attempt_number，最终只提交成功 attempt 的结果。
每个 TableBlockWriter 实例对应一个 block_number + attempt_number 对，会在内存中缓存该 Block 的所有批次，直到 commit() 或 close() 上传完整的 Arrow IPC Stream。大任务应拆为多个 Block。
open_block_writer 时 SDK 会自动预留一次上传 quota，调用方无需配置或传递 token。若无法完成 quota 预留，open_block_writer 会直接失败。
兼容模式必须调用 commit(block_results=...) 提交，无参 commit() 会报错。
BatchCompatibleOptions 的 max_field_size 设置后必须至少为 1024 字节， dynamic_partition_limit 使用 -1 表示服务端默认。
可通过 max_block_number 获取 Block 上限，合法范围为 [0, max_block_number)。

跨进程读写

MaxStorage 的会话 ID 是服务端分配的全局标识，可以在进程间传递。这使得 "一个进程创建会话、多个进程并行读写" 的架构成为可能。

除了通过 session_id 让工作进程重载会话，还可以把会话导出为会话描述符（见会话描述符（SessionDescriptor）），随任务分发后在本地恢复，全过程不需要创建或重载会话的请求。

跨进程读取

主进程创建读会话，将 session_id 和分片信息分发给多个工作进程。每个工作进程通过 session_id 重载会话，独立读取各自负责的分片：

# ---- 主进程：创建读会话，分发分片 ----
from odps.maxstorage import MaxStorageClient

client = MaxStorageClient(odps)
read_session = client.create_table_read_session("your_table")

session_id = read_session.id          # 传递给工作进程
splits = read_session.splits          # 分片列表，每个分片可独立读取

# 将 session_id 和 splits 分发给工作进程（例如通过 multiprocessing.Queue、
# Redis、文件等）

# ---- 工作进程：重载会话，读取指定分片 ----
from odps.maxstorage import MaxStorageClient

client = MaxStorageClient(odps)

# 通过 session_id 重载已有会话（不会创建新会话）
read_session = client.create_table_read_session("your_table", session_id=session_id)

# 每个工作进程只读取自己负责的分片
for split in my_splits:
    reader = read_session.open_arrow_reader(split)
    while True:
        batch = reader.read()
        if batch is None:
            break
        process(batch)
    reader.close()

跨进程写入

主进程创建写会话，将 session_id 分发给多个工作进程。每个工作进程通过 session_id 重载会话并以 resume=True 创建写流，独立写入数据。所有工作进程完成后，主进程提交会话：

# ---- 主进程：创建写会话 ----
from odps.maxstorage import MaxStorageClient, WriteMode

client = MaxStorageClient(odps)
write_session = client.create_table_write_session("your_table")
session_id = write_session.id          # 传递给工作进程

# 等待所有工作进程完成写入...

# 主进程提交会话（只需 session_id 即可重载并提交）
write_session = client.create_table_write_session("your_table", session_id=session_id)
write_session.commit()

# ---- 工作进程：重载会话，resume 写入 ----
from odps.maxstorage import MaxStorageClient

client = MaxStorageClient(odps)
write_session = client.create_table_write_session("your_table", session_id=session_id)

# resume=True 调用 getWriteStream 获取已有写流的状态
writer = write_session.open_arrow_writer(stream_id=my_stream_id, resume=True)

writer.write_batch(my_batch)
writer.close()   # 关闭写流（不会提交会话）

# 工作进程只需关闭写流，提交由主进程统一完成

每个工作进程使用不同的 ``stream_id``（"0"、"1"、"2"……）以并行写入同一会话。
- 工作进程的 writer.close() 仅关闭写流，不会提交数据。必须由主进程调用 write_session.commit() 才能使数据可见。
- 若工作进程异常退出，主进程可调用 write_session.abort() 丢弃所有已上传数据。
- 写会话的 session_id 重载时会自动获取最新的 route_token，确保请求路由到正确的服务端节点。

会话描述符（SessionDescriptor）

会话描述符是会话状态的一份不可变 JSON 描述，由 SessionDescriptor 表示。主进程用 to_descriptor() 或 to_descriptor() 导出，经任意应用通道（队列、文件、对象存储等）传给工作进程；工作进程用 from_json() 解析后，通过 create_table_read_session(descriptor=...) 或 create_table_write_session(descriptor=...) 在本地绑定会话。恢复过程不会创建或重载远端会话：描述符就是 schema、分片计划、路由 token 等会话状态的唯一来源，工作进程仍使用自己的客户端与凭据。

描述符的 kind 是 SessionKind 枚举：BATCH_READ、 INCREMENTAL_READ 和 BATCH_WRITE，在 JSON 中分别为 batch-read、 incremental-read 和 batch-write。SessionDescriptor.create 同时接受枚举成员与对应的字符串值。

描述符本身不含认证信息（密钥、签名），但包含会话 ID 与路由 token，应按任务内部数据对待，不要写入日志或对外公开。from_json() 会校验 version、provider、 kind、表标识以及各分片模式的必填字段，并丢弃未知字段，因此服务端新增字段不会破坏旧版本 SDK 的解析。

导出与恢复读会话

主进程导出描述符并随任务下发；工作进程恢复会话时在本地重建分片计划，不发起任何会话请求：

# ---- 主进程：导出描述符 ----
from odps.maxstorage import MaxStorageClient

client = MaxStorageClient(odps)
read_session = client.create_table_read_session("your_table")

descriptor_json = read_session.to_descriptor().to_json()   # 随任务下发

# ---- 工作进程：本地恢复会话，读取分片 ----
from odps.maxstorage import MaxStorageClient, SessionDescriptor

client = MaxStorageClient(odps)
descriptor = SessionDescriptor.from_json(descriptor_json)

# 不会创建或重载会话，分片列表由描述符中的分片计划在本地重建
read_session = client.create_table_read_session(descriptor=descriptor)

for split in read_session.splits:
    reader = read_session.open_arrow_reader(split)
    while True:
        batch = reader.read()
        if batch is None:
            break
        process(batch)
    reader.close()

备注

ROW_OFFSET 分片模式下描述符会记录 rowsPerSplit；导入时若缺少该值且记录数非零，from_descriptor 会抛出 ValueError。
增量读会话导出的描述符保留创建请求中的增量读取选项（服务端可能不回显），恢复后仍是相同的增量读取范围。

导出与恢复写会话

只有兼容模式（WriteMode.BATCH_COMPATIBLE）的写会话支持描述符。主进程导出后把 Block 写入交给工作进程，最终仍由主进程提交会话：

# ---- 主进程：导出描述符 ----
from odps.maxstorage import MaxStorageClient, WriteMode

client = MaxStorageClient(odps)
write_session = client.create_table_write_session(
    "your_table",
    write_mode=WriteMode.BATCH_COMPATIBLE,
)
descriptor_json = write_session.to_descriptor().to_json()   # 随任务下发

# ---- 工作进程：本地恢复会话，写入并提交 Block ----
from odps.maxstorage import MaxStorageClient, SessionDescriptor

client = MaxStorageClient(odps)
descriptor = SessionDescriptor.from_json(descriptor_json)
write_session = client.create_table_write_session(descriptor=descriptor)

writer = write_session.open_block_writer(my_block, 0)
batch = pa.RecordBatch.from_arrays(
    [pa.array([1, 2, 3], type=pa.int64())],
    schema=writer.schema,
)
writer.write_batch(batch)
result = writer.commit()     # BlockWriteResult，回传给主进程
write_session.close()        # 只关闭本地句柄，不会终止全局会话

# ---- 主进程：汇总所有 Block 结果后提交会话 ----
write_session.commit(block_results=results)

备注

其他写入模式调用 to_descriptor() 会抛出 NotImplementedError。
恢复出的写会话只具备 Block 写入权限（descriptor_only 为 True）：可以打开 Block Writer，但调用 commit() 或 abort() 会抛出 StorageClientError；会话的提交与终止始终由主进程负责。
若会话已发起过提交，或最近一次观测到的会话状态不再是 NORMAL，恢复出的会话会拒绝打开新的 Block Writer。
会话提交成功时描述符状态为 COMMITTED；会话已确认终止（abort() 成功）时为 ABORTED；句柄已关闭但结果未知时为 UNKNOWN。
descriptor 不能与 table 或 session_id 同时传入，否则会抛出 ValueError。

Exactly-Once 模式

通过 exactly_once_mode=True 启用精确一次语义：

writer = write_session.open_arrow_writer(
    stream_id="0",
    exactly_once_mode=True,
)
# 写入时每次 flush 携带当前 row_offset，服务端返回新的 ExactlyOnceRowOffset
writer.write_batch(batch)
writer.flush()

# 恢复时通过 get_row_offset() 获取最新偏移
writer = write_session.open_arrow_writer(
    stream_id="0",
    exactly_once_mode=True,
    resume=True,
)
offset = writer.get_row_offset()

Blob 读写

什么是 Blob

**Blob**（Binary Large Object）是 MaxCompute 中用于存储二进制数据的列类型（BLOB），适用于存储图片、音频、视频、文档等非结构化数据。在 MaxStorage 中，BLOB 列的值在 Arrow 层面以 bytes``（或 ``pa.binary()）表示。

什么是 Blob 引用

由于单条 Blob 数据可能很大，MaxStorage 不会把原始二进制内容直接内嵌在 Arrow 批次中写入存储层。而是采用**两阶段写入**：

上传 Blob 数据：通过 write_blob_batch / write_blob_stream 将原始二进制数据上传到服务端的 Blob 存储区，服务端返回一个 **blob 引用**（blob reference）。
写入引用：将该 blob 引用作为 bytes 填入 Arrow 批次中 BLOB 列对应的位置，随普通数据一起 write_batch 提交。服务端根据引用还原出实际的二进制内容。

**blob 引用**是一个不透明的 bytes``（或 ``str）值，由服务端在上传时生成，其内部编码对客户端不可见。客户端只需将其原样写入 BLOB 列即可，**不要**尝试解码、修改或拼接引用。读取时，BLOB 列返回的同样是 blob 引用；需要调用 BlobManager.read_blobs 传入引用才能取回真正的二进制数据。

简而言之：BLOB 列里存的是引用，不是原始二进制内容；引用是上传后拿到的 ``bytes``，用它换回原始数据靠 ``read_blobs``。

MaxStorage 不提供 Blob 包装类——blob 引用就是原始的 bytes/str，这使得它与 Arrow / pandas 等生态无缝衔接。

MaxStorage 提供两种 BLOB 写入模式：

**手动上传**（见下文）：显式调用 write_blob_batch / write_blob_stream 上传，拿到引用后自行填入 BLOB 列。适用于需要精确控制上传时机或元数据的场景。
**自动上传**（见自动上传）：创建 writer 时指定 auto_upload_blobs=True，直接在 BLOB 列传入 bytes / 文件对象， writer 自动上传并替换为引用。适用于无需手动管理引用的场景。

手动上传

在手动上传模式下，你显式调用 write_blob_batch / write_blob_stream 上传 BLOB 数据，拿到引用后自行填入 Arrow 批次的 BLOB 列。此模式适用于需要精确控制上传时机、批量大小或元数据的场景。创建 writer 时无需 auto_upload_blobs （默认即为普通 writer，BLOB 列接受引用 bytes）。

build_blob_write_item 的 data 参数接受 bytes、bytearray 或任意可读的文件对象（file-like，需实现 read/seek/tell）。传入文件对象时支持流式读取，无需预先将大 Blob 全量加载到内存：

from io import BytesIO

# bytes / bytearray —— 直接传入
item1 = writer.build_blob_write_item(b"image-bytes", column_name="img")
item2 = writer.build_blob_write_item(bytearray(b"more-bytes"), column_name="img")

# 文件对象 —— 支持流式读取，避免大 Blob 全量加载
item3 = writer.build_blob_write_item(open("large.jpg", "rb"), column_name="img")
item4 = writer.build_blob_write_item(BytesIO(b"in-memory-stream"), column_name="img")

import pyarrow as pa

write_session = client.create_table_write_session("your_table")
writer = write_session.open_arrow_writer(stream_id="0")

# 1. 批量上传 Blob，拿到引用（resp.blob_references 已是 list[bytes]）
items = [
    writer.build_blob_write_item(b"image1-bytes", column_name="img"),
    writer.build_blob_write_item(b"image2-bytes", column_name="img"),
]
resp = writer.write_blob_batch(items)
refs = resp.blob_references  # list[bytes]

# 2. 将引用写入 BLOB 列，随普通数据一起提交
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array(refs, pa.binary())],
    schema=pa.schema([("id", pa.int64()), ("img", pa.binary())]),
)
writer.write_batch(batch)

writer.close()
write_session.commit()

流式上传单个 Blob：

import pyarrow as pa

# 1. 流式上传单个 Blob，拿到引用（resp.blob_reference 已是 bytes）
blob_writer = writer.write_blob_stream(column_name="img")
with open("image.jpg", "rb") as f:
    while True:
        chunk = f.read(65536)
        if not chunk:
            break
        blob_writer.write(chunk)
resp = blob_writer.finish()
ref = resp.blob_reference  # bytes

# 2. 将引用写入 BLOB 列
batch = pa.RecordBatch.from_arrays(
    [pa.array([0], pa.int64()), pa.array([ref], pa.binary())],
    schema=pa.schema([("id", pa.int64()), ("img", pa.binary())]),
)
writer.write_batch(batch)

writer.close()
write_session.commit()

自动上传

在自动上传模式下，创建 writer 时指定 auto_upload_blobs=True，得到 TableArrowBlobUploadWriter。此时 BLOB 列可直接传入 bytes / 文件对象， writer 在 write_batch 或 rw.write() 时自动批量上传每个 BLOB 单元格并替换为引用。此模式适用于无需手动管理引用的场景。

**Arrow API 写入**（顶层 BLOB）：

import pyarrow as pa

write_session = client.create_table_write_session("blob_table")
writer = write_session.open_arrow_writer(stream_id="0", auto_upload_blobs=True)

# BLOB 列直接传 bytes，writer 自动上传
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array([b"img0", b"img1"], pa.binary())],
    schema=pa.schema([("id", pa.int64()), ("img", pa.binary())]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

**Record API 写入**（无需手动构造 Arrow 批次）：

from io import BytesIO
from odps.models import Record

write_session = client.create_table_write_session("blob_table")
writer = write_session.open_arrow_writer(stream_id="0", auto_upload_blobs=True)
rw = writer.get_as_record_writer()

# BLOB 列传 bytes / BytesIO / 文件对象，writer 自动批量上传
rw.write(Record(columns=["id", "img"], values=[1, b"raw-bytes"]))
rw.write(Record(columns=["id", "img"], values=[2, BytesIO(b"file-like")]))
rw.close()
write_session.commit()

传入文件对象时，writer 会读取其全部内容并**立即关闭**该文件句柄，因此调用方无需手动关闭。

**嵌套 BLOB 写入**（ARRAY<BLOB>，需 API v3）：

from io import BytesIO

# 嵌套 ARRAY<BLOB> 表：a BIGINT, b ARRAY<BLOB>
client_v3 = MaxStorageClient(odps, api_version="3")
write_session = client_v3.create_table_write_session("nested_blob_table")
writer = write_session.open_arrow_writer(stream_id="0", auto_upload_blobs=True)
rw = writer.get_as_record_writer()

# b 列传 list[bytes]，每个元素自动上传
rw.write([0, [b"blob_0_a", b"blob_0_b"]])
rw.write([1, [BytesIO(b"blob_1_a"), b"blob_1_b", BytesIO(b"blob_1_c")]])
rw.close()
write_session.commit()

备注

auto_upload_blobs=True 时返回 TableArrowBlobUploadWriter，否则返回普通 TableArrowWriter`（BLOB 列必须已包含引用 ``bytes`）。两种 writer 都支持 get_as_record_writer() 和手动上传辅助方法（build_blob_write_item / write_blob_stream / write_blob_batch）。传入文件对象（file-like）写入 BLOB 时，writer 默认**不会**关闭这些文件句柄；若需自动关闭，创建 writer 时指定 auto_close_files=True。该选项对 Arrow API（write_batch）、手动批量上传（write_blob_batch）与 Record API（get_as_record_writer）均生效。

读取 Blob

通过 open_blob_manager() 创建 BlobManager，再调用 read_blobs / read_blob 下载：

blob_manager = client.open_blob_manager("your_table")

# 读取多个 Blob，返回 BlobDataIterator
iterator = blob_manager.read_blobs([ref1, ref2])
for record in iterator:
    print(len(record.data))  # record.data 为 bytes

# 流式读取（避免缓冲整个 Blob）
stream_reader = blob_manager.read_blobs([ref1], stream=True)
while True:
    chunk = stream_reader.read(4096)
    if not chunk:
        break
    # 处理 chunk

# 读取单个 Blob，返回 file-like 对象
fp = blob_manager.read_blob(ref1)
data = fp.read()

通过 Record API 读取时，BLOB 列返回引用，需 BlobManager 下载：

# 顶层 BLOB
read_session = client.create_table_read_session("blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client.open_blob_manager("blob_table")

for record in reader.get_as_record_reader():
    ref = record[1]              # bytes —— blob 引用
    data = next(blob_manager.read_blobs([ref])).data
    print(record[0], len(data))  # a, len(b)

# 嵌套 ARRAY<BLOB>
read_session = client_v3.create_table_read_session("nested_blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client_v3.open_blob_manager("nested_blob_table")

for record in reader.get_as_record_reader():
    ref_list = record[1]         # list[bytes] —— 引用列表
    blobs = [b.data for b in blob_manager.read_blobs(ref_list)]
    print(record[0], [len(b) for b in blobs])

嵌套 Blob

当 BLOB 列嵌套在复杂类型（ARRAY、STRUCT、MAP）中时，需要通过 find_all_blob_column_ids() 解析出嵌套列的服务端列 ID（dot-path 形式如 b.element）。该特性仅支持 API 版本 3 及以上（创建客户端时指定 api_version="3"）。

BLOB 列在 Python / Arrow 中的表示取决于所在复杂类型。下表列出常见组合：

备注

dot-path 中 .element 表示 ARRAY 的元素层，.value 表示 MAP 的值层， .<field_name> 表示 STRUCT 的字段层。多层嵌套时按从外到内的顺序拼接。

ARRAY<BLOB> 示例

表结构：a BIGINT, b ARRAY<BLOB>。列 b 的每个元素是一个独立的 Blob。

**手动上传**（先批量上传，再将引用写入 Arrow）：

client_v3 = MaxStorageClient(odps, api_version="3")
write_session = client_v3.create_table_write_session("nested_blob_table")
writer = write_session.open_arrow_writer(stream_id="0")

# 嵌套列名：array<blob> 列 b → "b.element"
# 批量上传两行各 2~3 个 blob（resp.blob_references 已是 list[bytes]）
items = [
    writer.build_blob_write_item(b"row0_blob0", column_name="b.element"),
    writer.build_blob_write_item(b"row0_blob1", column_name="b.element"),
    writer.build_blob_write_item(b"row1_blob0", column_name="b.element"),
]
resp = writer.write_blob_batch(items)
refs = resp.blob_references  # list[bytes]

# 将引用按行组织成 list[bytes]，写入 Arrow list(binary) 列
row_refs = [refs[0:2], refs[2:3]]
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array(row_refs, pa.list_(pa.binary()))],
    schema=pa.schema([("a", pa.int64()), ("b", pa.list_(pa.binary()))]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

读取时，b 列返回 ``list[bytes]``（每个元素是一个引用），逐个下载即可：

read_session = client_v3.create_table_read_session("nested_blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client_v3.open_blob_manager("nested_blob_table")

for batch in reader:
    for a_val, blob_ref_list in zip(batch.column(0).to_pylist(),
                                    batch.column(1).to_pylist()):
        # blob_ref_list 是 list[bytes]（引用列表）
        blobs = [b.data for b in blob_manager.read_blobs(blob_ref_list)]
        print(a_val, [len(b) for b in blobs])

STRUCT<f: BLOB> 示例

表结构：id BIGINT, s STRUCT<f: BLOB>。列 s 是一个结构体，字段 f 是 Blob。

items = [writer.build_blob_write_item(b"struct_blob_0", column_name="s.f")]
resp = writer.write_blob_batch(items)
ref = resp.blob_references[0]  # bytes

# Arrow 行：{"f": <ref bytes>}
batch = pa.RecordBatch.from_arrays(
    [pa.array([0], pa.int64()),
     pa.array([{"f": ref}], pa.struct([("f", pa.binary())]))],
    schema=pa.schema([("id", pa.int64()),
                      ("s", pa.struct([("f", pa.binary())]))]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

读取时，s 列返回 dict，其 "f" 字段是引用：

for batch in reader:
    for id_val, s_val in zip(batch.column(0).to_pylist(),
                             batch.column(1).to_pylist()):
        ref = s_val["f"]
        data = next(blob_manager.read_blobs([ref])).data
        print(id_val, len(data))

Blob 元数据

每个 Blob 可携带两条元数据：MIME 类型**（``mime_type``）和**自定义文件名 （custom_file_name，仅 API v3）。写入时通过 BlobWriteItem 设置，读取时从 BlobRecord / BlobStreamReader 取回。

完整流程：上传带元数据的 Blob → 写入引用 → 读取引用 → 下载并验证元数据。

from odps.maxstorage import MaxStorageClient

client_v3 = MaxStorageClient(odps, api_version="3")

# --- 写入：上传带元数据的 Blob ---
write_session = client_v3.create_table_write_session("blob_table")
writer = write_session.open_arrow_writer(stream_id="0")

items = [
    writer.build_blob_write_item(
        b"\x89PNG image data",
        column_name="b",
        mime_type="image/png",
        custom_file_name="photo.png",
    ),
    writer.build_blob_write_item(
        b'{"key": "value"}',
        column_name="b",
        mime_type="application/json",
        custom_file_name="config.json",
    ),
]
resp = writer.write_blob_batch(items)
refs = resp.blob_references  # list[bytes]

# 将引用写入 Arrow 批次
batch = pa.RecordBatch.from_arrays(
    [pa.array([0, 1], pa.int64()), pa.array(refs, pa.binary())],
    schema=pa.schema([("a", pa.int64()), ("b", pa.binary())]),
)
writer.write_batch(batch)
writer.close()
write_session.commit()

# --- 读取：下载 Blob 并取回元数据 ---
read_session = client_v3.create_table_read_session("blob_table")
reader = read_session.open_arrow_reader(read_session.splits[0])
blob_manager = client_v3.open_blob_manager("blob_table")

for batch in reader:
    for a_val, ref in zip(batch.column(0).to_pylist(),
                          batch.column(1).to_pylist()):
        # 方式一：BlobRecord（一次性读取，含元数据）
        record = next(blob_manager.read_blobs([ref]))
        print(record.data)              # bytes —— 原始内容
        print(record.mime_type)         # "image/png" / "application/json"
        print(record.custom_file_name)  # "photo.png" / "config.json"

        # 方式二：BlobStreamReader（流式读取，含元数据）
        stream_reader = blob_manager.read_blobs([ref], stream=True)
        print(stream_reader.mime_type)        # 元数据在读取前即可获取
        print(stream_reader.custom_file_name)
        while True:
            chunk = stream_reader.read(4096)
            if not chunk:
                break
            # 处理 chunk
        stream_reader.next()  # 前进到下一个 Blob（若有）

备注

custom_file_name 仅在 API v3 及以上可用；v2 客户端上传时该字段被静默忽略，下载时 BlobRecord.custom_file_name 始终为 None。 mime_type 在 v2 / v3 均可用。

逐 Blob 元数据回调

上面的 "手动上传" 需要手动调用 write_blob_batch 上传 Blob 并在 BlobWriteItem 上逐个设置元数据。而在**自动上传模式**下（即直接把 BLOB 原始数据放入 Arrow 批次 write_batch，或通过 Record API rw.write() 写入）， writer 自动批量上传每个 BLOB 单元格。此时可通过 blob_metadata_callback 回调为**每一个 BLOB 单元格单独指定** mime_type 和 custom_file_name。

回调签名：callback(row_index, column_name, blob_data) -> (mime_type, custom_file_name) | None

row_index：当前行索引（从 0 开始）。
column_name：BLOB 列的 dot-path（顶层列为列名如 "b"，嵌套列为 "b.element"、"s.f"）。
blob_data：原始 BLOB 数据。Record API 路径下为用户传入的原始值（bytes / 文件对象），回调在文件被读取为 bytes 之前触发；Arrow API 路径下为已反序列化的 bytes。

备注

回调**对每个 BLOB 单元格恰好调用一次**。
None 值和已有的 blob 引用（str）不会触发回调——它们不会被上传。
回调返回 None 时，回退到 open_arrow_writer 传入的 blob_mime_type / blob_custom_file_name 会话级默认值（也可单独使用，无需回调）；若默认值也为 None，则该 Blob 不携带元数据。
流式上传（write_blob_stream）的 wire 协议不包含 framing header，因此流式路径**不支持**元数据。

示例（从文件名推导 custom_file_name）：

import os
from odps.models import Record

write_session = client.create_table_write_session("blob_table")

def metadata_fn(row_index, column_name, blob_data):
    # blob_data 是用户传入的原始文件对象，可用其 .name 推导文件名
    name = os.path.basename(getattr(blob_data, "name", f"blob_{row_index}"))
    return None, name

writer = write_session.open_arrow_writer(
    stream_id="0",
    auto_upload_blobs=True,
    blob_metadata_callback=metadata_fn,
)
rw = writer.get_as_record_writer()

rw.write(Record(columns=["a", "b"], values=[1, open("photo.jpg", "rb")]))
rw.write(Record(columns=["a", "b"], values=[2, open("doc.pdf", "rb")]))
rw.close()
write_session.commit()
# 两个 Blob 的 custom_file_name 分别为 "photo.jpg"、"doc.pdf"