Realtime Compute for Apache Flink:Démarrage rapide pour l'analyse de données en temps réel avec les grands modèles de langage

Informations générales

Le service Flink AI propose des grands modèles de langage (LLM) intégrés et gérés. Vous n'avez pas besoin de demander de clé API. Référencez directement les modèles intégrés dans vos jobs Flink SQL pour activer l'inférence IA et la vectorisation en streaming. Les sections suivantes décrivent deux cas d'utilisation principaux de ces modèles :

• Modèle chat/completions Un modèle chat/completions est un LLM basé sur la génération de dialogues et la compréhension du texte. Il est largement utilisé dans des scénarios tels que l'analyse des sentiments, la reconnaissance d'intention et les systèmes de questions-réponses.

  • Analyse des sentiments : classe en temps réel les sentiments exprimés dans les commentaires sur les réseaux sociaux de votre entreprise, afin d'identifier si le sentiment de l'utilisateur est positif, négatif ou neutre.

  • Service client intelligent : permet des interactions en langage naturel pour les systèmes de service client intelligents, grâce aux capacités de génération de dialogue.

  • Modération de contenu : détecte automatiquement les contenus sensibles ou les violations de politique dans le texte, pour des audits de sécurité de contenu plus efficaces.

• Modèle embedding Un modèle embedding convertit le texte en représentations vectorielles de haute dimension. Les applications courantes incluent la recherche sémantique, les systèmes de recommandation et la construction de graphes de connaissances.

  • Recherche sémantique : permet une recherche sémantique basée sur la pertinence en vectorisant les descriptions de produits ou les requêtes des utilisateurs.

  • Systèmes de recommandation : utilise la vectorisation du texte pour découvrir des associations entre les intérêts des utilisateurs et les caractéristiques des produits, améliorant ainsi la précision des recommandations.

  • Graphe de connaissances : convertit le texte non structuré au format vectoriel pour simplifier l'extraction ultérieure de connaissances et la modélisation des relations.

Prérequis

• Vous avez activé un espace de travail Flink. Pour plus d'informations, consultez la section Activer Realtime Compute for Apache Flink.

• Vous avez activé le service Flink AI. Pour plus d'informations, consultez la section Service Flink AI (modèles intégrés).

Limites

• Nécessite Ververica Runtime (VVR) version 11.7 ou ultérieure.

• Le débit de l'opérateur ML_PREDICT est limité par la politique de limitation de débit de la plateforme de service de modèle. Lorsque la limite de trafic est atteinte, le job Flink peut subir une contre-pression ou redémarrer en raison de délais d'attente dépassés.

Étape 1 : Enregistrer un modèle intégré

Pour plus d'informations, consultez la section Paramètres du modèle.

CREATE MODEL ai_analyze_sentiment
INPUT (`input` STRING)
OUTPUT (`content` STRING)
WITH (
    'provider' = 'dashscope',
    'task' = 'chat/completions',
    'model' = 'qwen3.5-flash',
    'system-prompt' = 'Classify the text below into one of the following labels: [positive, negative, neutral, mixed]. Output only the label.'
);

CREATE MODEL embedding_model
INPUT (`input` STRING)
OUTPUT (`embeddings` ARRAY<FLOAT>)
WITH (
    'provider' = 'dashscope',
    'task' = 'embeddings',
    'model' = 'text-embedding-v4'
);

Étape 2 : Créer un job

Créez un brouillon pour un job de streaming SQL. Pour plus d'informations, consultez la section Jobs Flink SQL.

Étape 3 : Rédiger un job SQL pour l'analyse LLM

-- Create a temporary sink table.
CREATE TEMPORARY TABLE print_sink(
  id BIGINT,
  movie_name VARCHAR, 
  predict_label VARCHAR, 
  actual_label VARCHAR
) WITH (
  'connector' = 'print',   -- Use the print connector.
  'logger' = 'true'        -- Display the results in the console.
);

-- Create a temporary data view to construct test data.
-- | id | movie_name | comment   | actual_label |
-- | 1  | Her Story     | My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving. | POSITIVE |
-- | 2  | The Dumpling Queen   | Unremarkable.  | NEGATIVE |
CREATE TEMPORARY VIEW movie_comment(id, movie_name, user_comment, actual_label)
AS VALUES (1, 'Her Story', 'My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving.', 'positive'), (2, 'The Dumpling Queen', 'Unremarkable.', 'negative');

INSERT INTO print_sink
SELECT id, movie_name, content as predict_label, actual_label 
FROM ML_PREDICT(
  TABLE movie_comment, 
  MODEL ai_analyze_sentiment,  -- The registered Flink built-in text model.
  DESCRIPTOR(user_comment));   

-- Create a temporary sink table named milvus_sink.
CREATE TEMPORARY TABLE milvus_sink
(
    id STRING,
    movie_name STRING,
    user_comment STRING,
    embeddings ARRAY<FLOAT>,
    PRIMARY KEY (id) NOT ENFORCED
)
WITH (
    'connector' = 'milvus',
    'endpoint' = '<YOUR-ENDPOINT>',
    'port' = '<YOUR-PORT>',
    'userName' = '<YOUR-USERNAME>',
    'password' = '<YOUR-PASSWORD>',
    'databaseName' = 'default',
    'collectionName' = 'movie-comment-embeddings'
);

-- Create a temporary data view to construct test data.
-- | id | movie_name | comment |
-- | 1 | Her Story |My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving.|
-- | 2 | The Dumpling Queen | Unremarkable. |
CREATE TEMPORARY VIEW movie_comment(id, movie_name,  user_comment)
AS VALUES ('1', 'Her Story', 'My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving.'), ('2', 'The Dumpling Queen', 'Unremarkable.');

INSERT INTO
    milvus_sink
SELECT
    id,
    movie_name,
    user_comment,
    embeddings
FROM
    ML_PREDICT (
        TABLE movie_comment,
        MODEL embedding_model,  -- The registered Flink built-in embedding model.
        DESCRIPTOR (user_comment)
    );

Étape 4 : Déployer et démarrer le job

Déployez et démarrez le job. Pour plus d'informations, consultez la section Jobs Flink SQL.

Étape 5 : Consulter les résultats de l'analyse

Documentation connexe

• Instructions DDL (Data Definition Language) pour les modèles d'IA : Configurer un modèle

• Fonctions IA : ML_PREDICT

• Service de recherche vectorielle : Milvus (aperçu public)