Blob storage

Download and configure the JAR

To use blob features, download the paimon-ali-emr-spark-3.5-1-ali-29.1.jar attachment.

Upload the JAR to an accessible OSS path and configure the following parameters:

spark.emr.serverless.excludedModules    paimon
spark.emr.serverless.user.defined.jars  oss://my-bucket/jars/paimon-ali-emr-spark-3.5-1-ali-29.1.jar

Create a table

In Spark SQL, use the BINARY type to declare a blob column:

CREATE TABLE my_db.image_table (
    id BIGINT,
    name STRING,
    category STRING,
    image BINARY
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'image'
);

Managed blob also supports collection types. Use the blob-field table option to declare multiple managed blob fields:

CREATE TABLE my_db.gallery_table (
    id BIGINT,
    gallery ARRAY<BINARY>,
    renditions MAP<STRING, BINARY>
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'gallery,renditions'
);

Write data

Write binary data directly with SQL:

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

In a notebook, read an image from OSS and write it to the blob table:

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

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

You can also use path_to_descriptor to read a file from OSS and write it as a managed blob:

INSERT INTO my_db.image_table VALUES
    (2, 'external.jpg', 'photo',
     sys.path_to_descriptor('oss://my-bucket/images/external.jpg'));

When Paimon writes the data, it reads the source file and copies the content to a .blob file managed by the current table.

Query data

Read only structured columns without loading blob data:

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

Read the blob content:

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

Set a dynamic parameter to read the blob descriptor:

SET spark.paimon.my_catalog.my_db.image_table.blob-as-descriptor=true;

SELECT id, name, sys.descriptor_to_string(image) AS image_descriptor
FROM my_db.image_table
WHERE id = 1;

RESET spark.paimon.my_catalog.my_db.image_table.blob-as-descriptor;

Write a Blob View reference

Use sys.blob_view in Spark to generate a Blob View reference from the upstream table, blob field, and _ROW_ID. Spark then writes the reference to a downstream field declared by the blob-view-field table option. This operation writes only the reference and does not copy the upstream blob content.

USE my_db;

CREATE TABLE image_view_table (
    id BIGINT,
    name STRING,
    image_ref BINARY
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-view-field' = 'image_ref'
);

INSERT INTO image_view_table
SELECT
    id,
    name,
    sys.blob_view(
        'my_catalog.my_db.image_table',
        'image',
        _ROW_ID
    )
FROM `image_table$row_tracking`;

SELECT id, name, length(image_ref) AS image_size
FROM image_view_table;

Pass the upstream table name, upstream blob field name, and upstream row _ROW_ID to sys.blob_view, in that order.

Update data

Use MERGE INTO in Spark to update single-value blobs, ARRAY<BLOB>, and MAP<K, BLOB>:

MERGE INTO my_db.image_table AS target
USING my_db.image_update_source AS source
ON target.id = source.id
WHEN MATCHED THEN
    UPDATE SET target.image = source.image;

For an append-only table without a primary key, configure upsert-key as the unique business key when you create the table. A write that matches an existing upsert-key updates the row. Otherwise, the write inserts a row. You cannot configure upsert-key together with a primary key:

CREATE TABLE my_db.image_upsert_table (
    id BIGINT,
    name STRING,
    category STRING,
    image BINARY
) TBLPROPERTIES (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'image',
    'upsert-key' = 'id'
);

-- The first write for id=1 inserts a row.
INSERT INTO my_db.image_upsert_table VALUES
    (1, 'sample.jpg', 'photo', X'89504E470D0A1A0A');

-- A subsequent write for id=1 updates the existing row.
INSERT INTO my_db.image_upsert_table VALUES
    (1, 'sample-new.jpg', 'photo', X'FFD8FFE000104A46');

SELECT id, name
FROM my_db.image_upsert_table
WHERE id = 1;
-- Returns: 1, sample-new.jpg

Download and configure the JAR

To use blob features, download the paimon-ali-vvr-11-vvp-1-ali-29.1-20260731.jar attachment. Its Catalog Factory and Table Factory identifiers are both paimon-1-ali-29-20260731.

In the Realtime Compute for Apache Flink console, upload the JAR in Data Management and access it through a custom catalog.

Add a catalog

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

Create a table and write data

In Flink SQL, use the BYTES type to declare a blob column:

CREATE TABLE my_catalog.my_db.image_table (
    id BIGINT,
    name STRING,
    category STRING,
    image BYTES
) WITH (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-field' = 'image'
);

INSERT INTO my_catalog.my_db.image_table VALUES
    (1, 'cat.jpg', 'photo', X'89504E470D0A1A0A'),
    (2, 'dog.jpg', 'photo',
     my_catalog.sys.path_to_descriptor(
         'oss://my-bucket/images/dog.jpg'
     ));

When you write the result of path_to_descriptor, Paimon reads the source file and writes its content to a .blob file managed by the current table.

Query data

Read only structured columns:

SELECT id, name, category
FROM my_catalog.my_db.image_table;

Read the blob content:

SELECT id, name
FROM my_catalog.my_db.image_table
/*+ OPTIONS('blob-as-descriptor'='false') */;

Read the blob descriptor:

SELECT
    id,
    name,
    my_catalog.sys.descriptor_to_string(image) AS image_descriptor
FROM my_catalog.my_db.image_table
/*+ OPTIONS('blob-as-descriptor'='true') */;

Use Blob View

Blob View lets a downstream table reference blob data in an upstream table without copying the content or creating new .blob files. Declare the downstream field by using the blob-view-field table option. The field stores only the reference, which DLF resolves to the upstream blob content when queried.

USE CATALOG my_catalog;
USE my_db;

CREATE TABLE image_view_table (
    id BIGINT,
    name STRING,
    image_ref BYTES
) WITH (
    'row-tracking.enabled' = 'true',
    'data-evolution.enabled' = 'true',
    'blob-view-field' = 'image_ref'
);

INSERT INTO image_view_table
SELECT
    id,
    name,
    sys.blob_view(
        'my_catalog.my_db.image_table',
        'image',
        _ROW_ID
    )
FROM `image_table$row_tracking`;

Pass the upstream table name, upstream blob field name, and upstream row _ROW_ID to sys.blob_view, in that order. By default, queries of image_ref return the referenced blob content.

Flink Data Evolution MERGE INTO does not currently support updating managed blob fields directly. Use Spark to update blob content.

Download and install PyPaimon

Download the pypaimon-1.5.dev20260727.tar.gz package.

Run the following command:

pip install pypaimon-1.5.dev20260727.tar.gz

Create a table and write blob data

import pyarrow as pa
from pypaimon import CatalogFactory, Schema

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',
    },
)

catalog.create_table('my_db.image_table', schema, True)
table = catalog.get_table('my_db.image_table')

write_builder = table.new_batch_write_builder()
writer = write_builder.new_write()
commit = write_builder.new_commit()

data = pa.Table.from_pydict({
    'id': [1],
    'name': ['sample.png'],
    'picture': [b'\x89PNG\r\n\x1a\n'],
}, schema=pa_schema)

writer.write_arrow(data)
commit.commit(writer.prepare_commit())
writer.close()
commit.close()

When you create a downstream Blob View table, set the blob-view-field option in the schema options to specify the reference field:

blob_view_pa_schema = pa.schema([
    ('id', pa.int64()),
    ('picture_ref', pa.large_binary()),
])

blob_view_schema = Schema.from_pyarrow_schema(
    blob_view_pa_schema,
    options={
        'row-tracking.enabled': 'true',
        'data-evolution.enabled': 'true',
        'blob-view-field': 'picture_ref',
    },
)

catalog.create_table(
    'my_db.image_view_table',
    blob_view_schema,
    True,
)

Read blob data

We recommend that you read blob data in batches:

read_builder = table.new_read_builder()
splits = read_builder.new_scan().plan().splits()
read = read_builder.new_read()

for batch in read.to_arrow_batch_reader(
        splits, blob_parallelism=16):
    for i in range(len(batch)):
        picture = batch['picture'][i].as_py()

Read only structured columns:

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)

For a large blob, set blob-as-descriptor=true and use a streaming API to read the data in chunks. This approach avoids loading the entire blob into memory at once.