DocsICP FilingConsole
Documentation
Home Page Realtime Compute for Apache Flink Streaming Compute Flink AI capabilities Scenarios Analyze e-commerce product images with the Flink Python DataFrame API

Analyze e-commerce product images with the Flink Python DataFrame API

Updated at:
Copy as MD
Product
My Favorites

This topic uses an e-commerce product image analysis example to show how to use the Flink Python DataFrame API to invoke a multimodal large language model. The example covers the full workflow from image reading and preprocessing to product description generation and text embedding. For more information about the DataFrame API, see Feature overview.

Prerequisites

  • A Flink workspace has been created. For more information, see Activate Realtime Compute for Apache Flink.

  • Flink AI Service has been enabled. For more information, see Flink AI Service (built-in models).

  • (Optional) Configure the local development environment. The Flink Python DataFrame API is available only in VVR 11.8 and later, and is not provided in open-source Flink. Install the VVR version of the PyFlink dependency:

    pip install "ververica-flink==11.8.0"
    Note

    ververica-flink is an API-Only package that provides only type definitions and interface declarations for local development. Jobs must still be submitted on the Realtime Compute for Apache Flink platform.

Data preparation

In this example, prepare test image files (1.jpg,2.jpg,3.jpg), and upload them to the Flink workspace through the file management feature. For upload steps, see File management.

Job development

Create a Python file (for example, pyflink_dataframe_example. py) and develop the job by following the steps below. The complete Python code is provided at the end of this topic.

Step 1: Register AI model providers

Use pf.set_model_provider to register two model providers: chat for image understanding, and embedding for converting product descriptions to text embeddings.

import pyflink.dataframe as pf

pf.set_model_provider(
    "chat",
    pf.OpenAICompatProvider(
        task="chat/completions",
        error_handling_strategy="RETRY",
        retry_num=3,
        retry_backoff_strategy="EXPONENTIAL",
    ),
)

pf.set_model_provider(
    "embedding",
    pf.OpenAICompatProvider(
        task="embeddings",
        error_handling_strategy="RETRY",
        retry_num=3,
        retry_backoff_strategy="EXPONENTIAL",
    ),
)

Step 2: Prepare input data

Use pf.from_dict to build the input DataFrame containing product IDs and their corresponding OSS paths.

OSS_PREFIX = "< your_oss_prefix>"

df = pf.from_dict(
    {
        "image_id": ["1", "2", "3"],
        "image_url": [
            f"{OSS_PREFIX}/1.jpg",
            f"{OSS_PREFIX}/2.jpg",
            f"{OSS_PREFIX}/3.jpg",
        ],
    }
)
Note

Choose the value of OSS_PREFIX based on the storage type of your Flink workspace. For more information, see File management:

  • If the storage type is an OSS bucket, OSS_PREFIX should be oss://< your-bound-OSS-bucket>/artifacts/namespaces/< namespace>.

  • If the storage type is fully managed storage, OSS_PREFIX should be oss://flink-fullymanaged-< workspace-id>/artifacts/namespaces/< namespace>.

Note

In production, you typically read data from external sources. The DataFrame API provides a variety of I/O functions for reading and writing external data. For more information, see I/O.

Step 3: Preprocess images with multimodal operators

This example uses multimodal operators to perform the following operations on the image URLs:

  1. Use fetch_content() to read image bytes from OSS.

  2. Use image. is_valid() to validate the image.

  3. Use image. decode() to decode the image bytes.

  4. Use image. size_filter() to check image dimensions.

  5. Use image. crop_black_border(), image. resize(), and image. encode(output="data_url") to produce a Data URL suitable for a multimodal model.

from pyflink.dataframe import col

preprocessed = (
    df. with_column("image_bytes", col("image_url"). fetch_content())
.with_column(
        "image_valid",
        col("image_bytes"). image. is_valid(
            pixel_limit=40_000_000,
            concurrency=16,
        ),
    )
.filter("image_valid")
.with_column(
        "image",
        col("image_bytes"). image. decode(
            on_error="null",
            mode="RGB",
            pixel_limit=40_000_000,
            concurrency=16,
        ),
    )
.drop_null(subset=["image"])
.with_columns(
        min_size_pass=col("image"). image. size_filter(
            min_w=256,
            min_h=256,
            concurrency=16,
        ),
        image_data=(
            col("image")
.image. crop_black_border(concurrency=16)
.image. resize(width=1024, height=1024, concurrency=16)
.image. encode(
                format="JPEG",
                quality=85,
                output="data_url",
                concurrency=16,
            )
        ),
    )
.filter("min_size_pass")
)

Step 4: Invoke AI functions

Pass the preprocessed image Data URL to llm. predict and use qwen3.6-plus to generate product descriptions. Then call llm. ai_embed with text-embedding-v4 to generate description embeddings.

result = (
    preprocessed. llm. predict(
        "image_data",
        provider="chat",
        model="qwen3.6-plus",
        output_type=pf.DataType. struct(
            {"image_description": pf.DataType. string()}
        ),
        content_type="IMAGE_URL",
        system_prompt=(
            "You are an e-commerce product analysis assistant. Generate an accurate and concise product description based on the product image; "
            "return only the description."
        ),
        temperature=0.2,
    )
.llm. ai_embed(
        "image_description",
        provider="embedding",
        model="text-embedding-v4",
    )
.select(
        "image_id",
        "image_description",
        "embedding",
    )
)

Step 5: Write results

Write the processed results to a Parquet file.

result. write_parquet(
    f"{OSS_PREFIX}/product_descriptions/"
)
Note

For detailed documentation of the DataFrame API, see API reference.

Deploy and run the job

  1. Deploy the Python job. For the deployment procedure, see Flink Python jobs.

  2. Configure the job:

    1. On the Operations > Jobs page, click the target job name.

    2. On the Deployment details tab, in the Basic configuration area, set Engine version to vvr-11.8-jdk11-flink-1.20.

  3. Run the job:

    1. On the Operations > Jobs page, click Start in the Actions column of the target job.

    2. Select Stateless start, click Start. For more information about starting jobs, see Start a job.

      After clicking Start, the job status becomes Running or Completed, indicating that the job is running correctly. If you deploy the Python test file provided in this topic, the job eventually finishes in the Completed state.

  4. After the job finishes, download the Parquet files under the product_descriptions/  directory from the File Management page to verify the output. The output contains the following columns:

    Column

    Type

    Description

    image_id

    STRING

    Product ID

    image_description

    STRING

    Product description generated by the model

    embedding

    ARRAY< FLOAT>

    Embedding vector of the description

Complete code

import pyflink.dataframe as pf
from pyflink.dataframe import col

OSS_PREFIX = "< your_oss_prefix>"

pf.set_model_provider(
    "chat",
    pf.OpenAICompatProvider(
        task="chat/completions",
        error_handling_strategy="RETRY",
        retry_num=3,
        retry_backoff_strategy="EXPONENTIAL",
    ),
)

pf.set_model_provider(
    "embedding",
    pf.OpenAICompatProvider(
        task="embeddings",
        error_handling_strategy="RETRY",
        retry_num=3,
        retry_backoff_strategy="EXPONENTIAL",
    ),
)

df = pf.from_dict(
    {
        "image_id": ["1", "2", "3"],
        "image_url": [
            f"{OSS_PREFIX}/1.jpg",
            f"{OSS_PREFIX}/2.jpg",
            f"{OSS_PREFIX}/3.jpg",
        ],
    }
)

preprocessed = (
    df. with_column("image_bytes", col("image_url"). fetch_content())
.with_column(
        "image_valid",
        col("image_bytes"). image. is_valid(
            pixel_limit=40_000_000,
            concurrency=16,
        ),
    )
.filter("image_valid")
.with_column(
        "image",
        col("image_bytes"). image. decode(
            on_error="null",
            mode="RGB",
            pixel_limit=40_000_000,
            concurrency=16,
        ),
    )
.drop_null(subset=["image"])
.with_columns(
        min_size_pass=col("image"). image. size_filter(
            min_w=256,
            min_h=256,
            concurrency=16,
        ),
        image_data=(
            col("image")
.image. crop_black_border(concurrency=16)
.image. resize(width=1024, height=1024, concurrency=16)
.image. encode(
                format="JPEG",
                quality=85,
                output="data_url",
                concurrency=16,
            )
        ),
    )
.filter("min_size_pass")
)

result = (
    preprocessed. llm. predict(
        "image_data",
        provider="chat",
        model="qwen3.6-plus",
        output_type=pf.DataType. struct(
            {"image_description": pf.DataType. string()}
        ),
        content_type="IMAGE_URL",
        system_prompt=(
            "You are an e-commerce product analysis assistant. Generate an accurate and concise product description based on the product image; "
            "return only the description."
        ),
        temperature=0.2,
    )
.llm. ai_embed(
        "image_description",
        provider="embedding",
        model="text-embedding-v4",
    )
.select(
        "image_id",
        "image_description",
        "embedding",
    )
)

result. write_parquet(
    f"{OSS_PREFIX}/product_descriptions/"
)
Previous：Invoke a Triton inference serviceNext：SQL AI functions