Realtime Compute for Apache Flink:Annotation intelligente de texte par IA basée sur YAML CDC

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
      '

{
    "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."
}    

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=()
}       

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         

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

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=()
}

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

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

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=()
}