AI intelligent text annotation Based on YAML CDC - Realtime Compute for Apache Flink

Solution architecture

Enterprises generate large volumes of unstructured text data every day, such as customer service tickets, user reviews, and complaint records. Traditional approaches rely on manual annotation or T+1 offline batch processing, which are costly, slow, and inconsistent.

By calling AI Functions in the Transform module of a CDC YAML job, data is annotated in real time as it flows from Kafka to the Paimon data lake, achieving sub-second streaming annotation.

This solution covers the following typical scenarios:

Customer complaint risk annotation: Uses AI_COMPLETE to assess complaint severity, detect exposure intent, and perform multi-dimensional risk annotation on customer service tickets.
Content classification annotation: Uses AI_CLASSIFY to automatically classify user reviews or feedback into predefined categories.
Sensitive information masking: Uses AI_MASK to identify and mask personally identifiable information (PII) in data before it enters the data lake.

Prerequisites

Realtime Compute for Apache Flink is activated and a workspace is created. For more information, see Activate Realtime Compute for Apache Flink.
Flink AI Service (built-in models) is activated.
Only Realtime Compute engine version VVR 11.9 (including preview editions) and later is supported.
A Kafka instance is created, and a topic with text data for annotation is prepared.

Limits

AI Functions can only be used in the projection expression of the Transform module in CDC YAML jobs.
The throughput of AI Functions is subject to the rate limits of Alibaba Cloud Model Studio. When the rate limit is reached, Flink jobs may experience backpressure or timeout restarts. For more information, see Rate limiting .

The following built-in Model Studio models are supported in the current version:

Model	Description
qwen3.6-flash	General text generation, recommended for annotation scenarios
qwen3.6-plus	Stronger reasoning capability
qwen3.5-flash	General text generation
qwen3.5-plus	Stronger reasoning capability
text-embedding-v4	Text vectorization

AI Function reference

The Transform module of CDC YAML supports the following AI Functions.

Function	Signature	Return fields	Description
AI_COMPLETE	`AI_COMPLETE('modelName', input, systemPrompt)`	`result`	General text generation
AI_CLASSIFY	`AI_CLASSIFY('modelName', input, labels)`	`category`, `confidence`	Text classification
AI_TRANSLATE	`AI_TRANSLATE('modelName', input, sourceLang, targetLang)`	`translated_text`, `detected_language`	Text translation
AI_SUMMARIZE	`AI_SUMMARIZE('modelName', input, maxLength)`	`summary`	Text summarization
AI_SENTIMENT	`AI_SENTIMENT('modelName', input)`	`score`, `label`, `confidence`	Sentiment analysis
AI_EXTRACT	`AI_EXTRACT('modelName', input, schema)`	`extracted_json`	Structured information extraction
AI_MASK	`AI_MASK('modelName', input, entities)`	`masked_text`, `detected_entities`	Sensitive information masking

Use ['field_name'] to access return fields in the projection expression. For example: AI_COMPLETE('mymodel', content, '')['result'].

Scenario 1: Customer complaint risk annotation

Use AI_COMPLETE to perform multi-dimensional risk annotation on customer service ticket text, including complaint severity level, exposure intent, and brand defamation tendency.

Job example

Note

Additional dependency files and configurations are required for deployment. For more information, see Deployment and startup.

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: <YOUR-KAFKA-BROKER>:9092
  topic: customer_complaints
  value.format: json
  scan.startup.mode: earliest-offset

# Output test results to console logs
sink:
  type: values

# sink:
#  type: paimon
#  catalog.properties.metastore: rest
#  catalog.properties.uri: <YOUR-DLF-ENDPOINT>
#  catalog.properties.warehouse: <YOUR-CATALOG-NAME>
#  catalog.properties.token.provider: dlf
#  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: customer_complaints
    projection: >
      comment,
      AI_COMPLETE('mymodel', comment, '')['result'] as risk_label

pipeline:
  parallelism: 2
  model:
  - name: mymodel
    type: openai-compatible
    model-name: qwen3.6-flash
    system-prompt: 'Prompt:
      #You are an e-commerce customer complaint risk assessment expert. Perform objective annotation based on user review content.
      #Annotation rules
      1. Complaint severity level (choose one): Normal / Dissatisfied / Angry
      2. Explicit exposure intent (yes/no): Yes / No
      3. Brand defamation tendency (yes/no): Yes / No
      #Output (strictly follow this structure, one item per line)
      Complaint severity: xxx
      Explicit exposure intent: xxx
      Brand defamation tendency: xxx
      Confidence: 0.xx
      Annotation basis: xxx
      '

Input and output examples

Input data (JSON format):

{
    "comment": "It broke less than a month after purchase, customer service kept telling me to wait, waited two weeks with no response. If this isn't resolved, I'll post it online to expose you."
}

Output data:

DataChangeEvent{
  tableId=customer_complaints,
  after=[
    The customer called to report that an air conditioner purchased in March 2024...,
    "Complaint severity: Angry
     Explicit exposure intent: Yes
     Brand defamation tendency: Yes
     Confidence: 0.95
     Annotation basis: If this isn't resolved, I'll post it online to expose you."
  ],
  op=INSERT, meta=()
}

Note

The data type of AI Function output fields is VARIANT, which facilitates accessing structured content in JSON format. If downstream consumers require a specific type (such as STRING or INT), use CAST for explicit conversion, for example CAST(risk_label AS STRING).

Scenario 2: Content classification annotation

Use AI_CLASSIFY to automatically classify user reviews or feedback text into predefined categories.

Job example

Note

Additional dependency files and configurations are required for deployment. For more information, see Deployment and startup.

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: <YOUR-KAFKA-BROKER>:9092
  topic: user_reviews
  value.format: json
  scan.startup.mode: earliest-offset

# Output test results to console logs
sink:
  type: values

# sink:
#  type: paimon
#  catalog.properties.metastore: rest
#  catalog.properties.uri: <YOUR-DLF-ENDPOINT>
#  catalog.properties.warehouse: <YOUR-CATALOG-NAME>
#  catalog.properties.token.provider: dlf
#  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: user_reviews
    projection: >
      content,
      AI_CLASSIFY('mymodel', content, 'ARRAY[''Product quality'',''After-sales service'',''Logistics issues'',''Price disputes'']')['category'] as category,
      AI_CLASSIFY('mymodel', content, 'ARRAY[''Product quality'',''After-sales service'',''Logistics issues'',''Price disputes'']')['confidence'] as confidence

pipeline:
  parallelism: 2
  model:
  - name: mymodel
    type: openai-compatible
    model-name: qwen3.6-flash

Input and output examples

Input data:

{"content": "The air conditioner is way too noisy. It buzzes as soon as it turns on, making it impossible to sleep."}

Output data:

DataChangeEvent{
  tableId=user_reviews,
  before=[],
  after=[The air conditioner is way too noisy. It buzzes as soon as it turns on, making it impossible to sleep., "Product quality", 0.95],
  op=INSERT, meta=()
}

Scenario 3: Sensitive information masking

Use AI_MASK to automatically identify and mask personally identifiable information (names, phone numbers, and addresses) in data entering the data lake.

Job example

Note

Additional dependency files and configurations are required for deployment. For more information, see Deployment and startup.

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: <YOUR-KAFKA-BROKER>:9092
  topic: user_feedbacks
  value.format: json
  scan.startup.mode: earliest-offset

# Output test results to console logs
sink:
  type: values

# sink:
#  type: paimon
#  catalog.properties.metastore: rest
#  catalog.properties.uri: <YOUR-DLF-ENDPOINT>
#  catalog.properties.warehouse: <YOUR-CATALOG-NAME>
#  catalog.properties.token.provider: dlf
#  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: user_feedbacks
    projection: >
      AI_MASK('mymodel', content, 'ARRAY[''name'',''phone number'',''ID number'',''address'']')['masked_text'] as masked_content,
      AI_MASK('mymodel', content, 'ARRAY[''name'',''phone number'',''ID number'',''address'']')['detected_entities'] as detected_entities

pipeline:
  parallelism: 2
  model:
  - name: mymodel
    type: openai-compatible
    model-name: qwen3.6-flash

Input and output examples

Input data:

{"content": "Customer Zhang San, phone number 13812345678, address 88 Jianguo Road, Chaoyang District, Beijing, reported that the refrigerator is not cooling."}

Output data:

DataChangeEvent{
  tableId=user_feedbacks,
  before=[],
  after=["Customer Zhang*, phone number 138****5678, address Chaoyang District, Beijing **, reported that the refrigerator is not cooling.", [
  {"type":"name","value":"Zhang San"},
  {"type":"phone number","value":"13812345678"},
  {"type":"address","value":"88 Jianguo Road, Chaoyang District, Beijing"}
  ]],
  op=INSERT, meta=()
}

Deployment and startup

Log on to the Realtime Compute console . Create a CDC YAML data ingestion job and paste the YAML configuration for your scenario into the editor. For more information, see Develop a Flink CDC data ingestion job .
Upload additional dependency files. In the Additional Dependencies section of the job draft, upload the following JAR package:

flink-cdc-pipeline-model-openai-compatible
Deploy and start the job.

Monitoring and rate limiting

In the current version, you can monitor the token usage of AI Functions through the Flink Web UI. In Flink Dashboard, click the target job and select the Metrics tab. In the Add Metric drop-down list, select the metric of the target operator (for example, ai_model_completion_tokens) to view the current metric value.