Blob存储

默认 Blob 存储

Blob 原始字节写入 DLF Paimon 管理的 .blob 文件，存储在表路径下，由 DLF 统一管理文件生命周期。

适用场景：直接写入图片、视频等二进制数据，由 DLF 统一管理存储和权限。

描述符引用存储（Descriptor Only）

配置 blob-descriptor-field 后，DLF Paimon 不创建 .blob 文件，而是将序列化的 BlobDescriptor（包含 URI、offset、length）直接存储在 Parquet 等数据文件中。

适用场景：多表关联场景。例如一张表存储 Blob 原始数据，另一张表通过描述符引用关联已有的 Blob 文件，避免重复存储。

建表

-- 创建包含 Blob 列的图片表
CREATE TABLE my_db.image_table (
    id INT,
    name STRING,
    category STRING,
    image BINARY
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'image'
);

Spark SQL 中使用 BINARY 类型，通过 blob-field 选项声明为 Blob 存储。

写入数据

通过 SQL 直接写入二进制数据：

INSERT INTO my_db.image_table VALUES (1, 'sample', 'photo', X'89504E470D0A1A0A');

通过 Notebook 从 OSS 读取图片写入 Blob 表（当前会话需具有对应 OSS 的访问权限）：

image_df = spark.read.format("binaryFile").load("oss://my_bucket/path/test.jpg")

image_df.selectExpr(
    "1 as id",
    "'test.jpg' as name",
    "'photo' as category",
    "content as image"
).writeTo("my_db.image_table").append()

查询数据

仅读取结构化列，不加载 Blob 数据：

SELECT id, name, category FROM my_db.image_table WHERE category = 'photo';

读取包含 Blob 的完整行：

SELECT id, name, length(image) as img_size FROM my_db.image_table WHERE id = 1;

在 Notebook 中展示图片：

from IPython.display import Image, display

row = spark.sql("SELECT image FROM my_db.image_table WHERE id = 1").first()
display(Image(data=row["image"]))

描述符模式写入

当数据已存储在 OSS 上时，可通过 path_to_descriptor 函数直接写入文件引用：

CREATE TABLE my_db.video_table (
    id INT,
    title STRING,
    video BINARY
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'video',
    'blob-descriptor-field' = 'video'
);

-- 引用 OSS 上已有的视频文件
INSERT INTO my_db.video_table VALUES
    (1, 'demo.mp4',  sys.path_to_descriptor('oss://my-bucket/videos/demo.mp4')),
    (2, 'intro.mp4', sys.path_to_descriptor('oss://my-bucket/videos/intro.mp4'));

描述符模式：关联已有 Blob 表

创建标注结果表，通过描述符引用原始图片，不重复存储：

CREATE TABLE my_db.annotation_table (
    annotation_id INT,
    image_id INT,
    label STRING,
    confidence DOUBLE,
    image BINARY
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'image',
    'blob-descriptor-field' = 'image'
);

从原始表读取描述符，写入关联表：

INSERT INTO my_db.annotation_table
SELECT
    row_number() OVER (ORDER BY t.id) as annotation_id,
    t.id as image_id,
    'cat' as label,
    0.95 as confidence,
    t.image  -- 写入的是 BlobDescriptor，指向原始表的 .blob 文件
FROM my_db.image_table /*+ OPTIONS('blob-as-descriptor'='true') */ t;

添加Catalog

自定义Catalog的type类型默认为paimon-multimodal-test，自定义type时，不可设置为paimon，会与原本的Connector冲突。

CREATE CATALOG my_catalog WITH (
  'type' = 'paimon-multimodal-test',
  'metastore' = 'rest',
  'token.provider' = 'dlf',
  'uri' = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com',
  'warehouse' = 'dlf_test'
);

建表与写入

Flink SQL中使用BYTES类型声明Blob列，通过blob-field选项指定为Blob存储。

-- 创建包含Blob列的图片表（默认Blob存储模式）
CREATE TABLE my_catalog.my_db.image_table (
    id INT,
    name STRING,
    category STRING,
    image BYTES
) WITH (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'image',
    'blob-as-descriptor' = 'true'
);

-- 写入数据
INSERT INTO my_catalog.my_db.image_table VALUES
    (1, 'cat.jpg',  'photo', my_catalog.sys.path_to_descriptor('oss://my-bucket/images/cat.jpg')),
    (2, 'dog.jpg',  'photo', my_catalog.sys.path_to_descriptor('oss://my-bucket/images/dog.jpg'));

查询数据

-- 仅读取结构化列，不加载 Blob 数据
SELECT id, name, category FROM my_catalog.my_db.image_table;

-- 读取Blob内容，获取文件大小
SELECT id, name, descriptor_to_string(image) as image_desc FROM my_catalog.my_db.image_table;

-- 以描述符形式读取（获取BlobDescriptor而非实际内容）
SELECT id, name, length(image) as img_size
FROM my_catalog.my_db.image_table /*+ OPTIONS('blob-as-descriptor'='false') */;

也可以通过 INSERT INTO ... SELECT 从其他已有表导入数据。

建表与写入 Blob 数据

import pyarrow as pa
from pypaimon import CatalogFactory, Schema

# 连接 DLF Catalog
catalog = CatalogFactory.create({
    'metastore': 'rest',
    'uri': 'https://${regionId}-vpc.dlf.aliyuncs.com',
    'warehouse': 'my_catalog',
    'token.provider': 'dlf',
    'dlf.access-key-id': '<AK>',
    'dlf.access-key-secret': '<SK>',
})

pa_schema = pa.schema([
    ('id', pa.int64()),
    ('name', pa.string()),
    ('picture', pa.large_binary())
])

schema = Schema.from_pyarrow_schema(
    pa_schema,
    options={
        'row-tracking.enabled': 'true',
        'data-evolution.enabled': 'true',
        'blob-field': 'picture',
    }
)
catalog.create_table('my_db.image_table', schema, True)
table = catalog.get_table('my_db.image_table')

# 写入 Blob 数据
write_builder = table.new_batch_write_builder()
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()

image_bytes = b'\x89PNG\r\n\x1a\n' + b'\x00' * 100

data = pa.Table.from_pydict({
    'id': [1],
    'name': ['sample.png'],
    'picture': [image_bytes],
}, schema=pa_schema)

table_write.write_arrow(data)
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()

从 OSS 读取图片写入：

from pypaimon.common.file_io import FileIO
from pypaimon.common.options import Options

oss_io = FileIO.get('oss://my-bucket/', Options({
    'fs.oss.accessKeyId': '<YOUR-AK>',
    'fs.oss.accessKeySecret': '<YOUR-SK>',
    'fs.oss.endpoint': 'https://oss-cn-hangzhou.aliyuncs.com',
}))
with oss_io.new_input_stream('oss://my-bucket/path/test.jpg') as f:
    image_bytes = f.read()

读取 Blob 数据

read_builder = table.new_read_builder()
splits = read_builder.new_scan().plan().splits()
result = read_builder.new_read().to_arrow(splits)

# 获取图片二进制数据
for i in range(result.num_rows):
    pic_bytes = result.column('picture')[i].as_py()
    with open(f'output_{i}.png', 'wb') as f:
        f.write(pic_bytes)

列裁剪：仅读取结构化列

read_builder = table.new_read_builder().with_projection(['id', 'name'])
splits = read_builder.new_scan().plan().splits()
result = read_builder.new_read().to_pandas(splits)

描述符模式：关联已有 Blob 表

from pypaimon.table.row.blob import BlobDescriptor

# 建表：video 列以描述符方式存储
pa_schema = pa.schema([
    ('id', pa.int32()),
    ('title', pa.string()),
    ('video', pa.large_binary()),
])
schema = Schema.from_pyarrow_schema(
    pa_schema,
    options={
        'row-tracking.enabled': 'true',
        'data-evolution.enabled': 'true',
        'blob-field': 'video',
        'blob-descriptor-field': 'video',
    }
)
catalog.create_table('my_db.video_table', schema, False)
table = catalog.get_table('my_db.video_table')

# 构造 BlobDescriptor（指向已存在的 OSS 文件）
video_path = 'oss://my-bucket/videos/demo.mp4'
video_size = 1024000  # 文件大小（字节），需提前获取
descriptor = BlobDescriptor(video_path, 0, video_size)

# 写入描述符
write_builder = table.new_batch_write_builder()
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()

table_write.write_arrow(pa.Table.from_pydict({
    'id': [1],
    'title': ['demo video'],
    'video': [descriptor.serialize()],
}, schema=pa_schema))

table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()

读取描述符并还原数据：

from pypaimon.table.row.blob import BlobDescriptor, Blob
from pypaimon.common.uri_reader import FileUriReader

read_builder = table.new_read_builder()
splits = read_builder.new_scan().plan().splits()
result = read_builder.new_read().to_arrow(splits)

for i in range(result.num_rows):
    video_bytes = result.column('video')[i].as_py()
    descriptor = BlobDescriptor.deserialize(video_bytes)
    print(f"URI: {descriptor.uri}, offset: {descriptor.offset}, length: {descriptor.length}")

    # 还原为实际数据
    uri_reader = FileUriReader(table.file_io)
    blob = Blob.from_descriptor(uri_reader, descriptor)
    data = blob.to_data()
    print(f"Blob data size: {len(data)} bytes")