Realtime Compute for Apache Flink:Connecteur Simple Message Queue (anciennement MNS)

Présentation

Par exemple, dans un pipeline de traitement d'images en temps réel, le connecteur SMQ détecte les fichiers nouvellement téléchargés dans un bucket OSS, la fonction FETCH_CONTENT télécharge l'image et l'intégration du modèle de langage IA réalise une analyse multimodale.

Prérequis

Avant de commencer, vérifiez que vous avez :

• Simple Message Queue (anciennement MNS) : Activation et autorisations. Pour un accès via le réseau interne, la file d'attente SMQ doit se trouver dans la même région que votre espace de travail Flink.

• (Pour un accès Internet public) Accès public activé pour votre espace de travail Flink et ajoutez ses adresses IP publiques à la liste d'autorisation de la file d'attente MNS. Pour plus de détails, consultez la section Contrôle d'accès.

Limites

• Le connecteur SMQ requiert Ververica Runtime (VVR) version 11.6.0 ou ultérieure.

• Pas de consommation basée sur l'offset : Contrairement à Apache Kafka, le connecteur SMQ ne prend pas en charge la consommation à partir d'un offset spécifique ni la recherche d'offset.

• Parallélisme fixe de 1 : Le connecteur SMQ garantit la sémantique exactly-once grâce à la déduplication côté connecteur et ne prend en charge qu'un parallélisme de 1.

• Points de contrôle requis : Le connecteur s'appuie sur les points de contrôle pour acquitter et supprimer les messages. Sans points de contrôle, les messages ne sont jamais supprimés, ce qui entraîne une consommation répétée infinie.

• Limite de taille du corps du message : Le corps d'un seul message MNS ne peut pas dépasser 64 Ko. Pour les messages plus volumineux, consultez la section Transmettre des messages de grande taille.

• Configuration du délai de visibilité : Définissez le délai de visibilité sur une valeur supérieure à l'intervalle des points de contrôle afin d'éviter la redistribution des messages.

• Aucun ordre strict : Les files d'attente standard MNS ne respectent pas strictement le principe Premier entré, premier sorti (FIFO). La redistribution des messages après expiration du délai peut modifier l'ordre de consommation.

Syntaxe

CREATE TABLE mns_source (
    data STRING
) WITH (
    'connector' = 'mns',
    'endpoint' = '${endpoint}',
    'region' = '${region}',
    'queueName' = '${queueName}',
    'accessKeyId' = '${accessKeyId}',
    'accessKeySecret' = '${accessKeySecret}',
    'format' = 'json',
    'batchSize' = '8',
    'pollingWaitTime' = '10s',
    'messageType' = 'RAW'
);

Options du connecteur

Lire les notifications d'événements OSS

Lorsque 'messageType' = 'OSS', le connecteur SMQ analyse automatiquement le JSON de notification d'événement OSS et mappe les champs à une structure de table plate. Les mappages de noms de champs suivants sont pris en charge (sans distinction entre majuscules et minuscules) :

Les paramètres de format JSON suivants contrôlent le comportement d'analyse :

• json.fail-on-missing-field : Indique si l'opération échoue lorsqu'un champ est manquant. Valeur par défaut : false.

• json.ignore-parse-errors : Indique si les erreurs d'analyse sont ignorées. Valeur par défaut : false.

• json.timestamp-format.standard : La norme de format d'horodatage. Valeur par défaut : ISO-8601.

• json.timestamp-format.pattern : Un modèle personnalisé pour le format d'horodatage.

Pour des descriptions détaillées des champs, consultez la section Notifications d'événements OSS.

Exemples

-- Create an MNS source table.
-- Field names must match the keys in the JSON message body.
CREATE TEMPORARY TABLE mns_source (
  `userId` BIGINT,
  `action` STRING,
  `timestamp` TIMESTAMP(3),
  `payload` STRING
) WITH (
  'connector' = 'mns',
  'endpoint' = 'http://your-account-id.mns.cn-hangzhou.aliyuncs.com',
  'region' = 'cn-hangzhou',
  'queueName' = 'my-events-queue',
  'accessKeyId' = 'your-ak',
  'accessKeySecret' = 'your-sk',
  'format' = 'json'
);

-- Create a print sink to test output.
CREATE TEMPORARY TABLE print_sink (
  `userId` BIGINT,
  `action` STRING,
  `timestamp` TIMESTAMP(3),
  `payload` STRING
) WITH (
  'connector' = 'print'
);

-- Consume messages and write to the sink.
INSERT INTO print_sink
SELECT userId, action, timestamp, payload
FROM mns_source;

-- Create an MNS source table to automatically parse OSS event notification JSON.
-- For field name mappings, see the "Read OSS event notifications" section.
CREATE TEMPORARY TABLE oss_event_source (
  `eventName` STRING,
  `eventSource` STRING,
  `eventTime` TIMESTAMP(3),
  `region` STRING,
  `ossBucketName` STRING,
  `ossObjectKey` STRING,
  `ossObjectSize` BIGINT,
  `responseElementsRequestId` STRING,
  `userIdentityPrincipalId` STRING
) WITH (
  'connector' = 'mns',
  'endpoint' = 'http://123456789.mns.cn-hangzhou.aliyuncs.com',
  'region' = 'cn-hangzhou',
  'queueName' = 'oss-events-queue',
  'accessKeyId' = '${secret_values.ak_id}',
  'accessKeySecret' = '${secret_values.ak_secret}',
  'format' = 'json',
  'messageType' = 'OSS'
);

-- Create a print sink to test output.
CREATE TEMPORARY TABLE print_sink (
  `eventName` STRING,
  `ossBucketName` STRING,
  `ossObjectKey` STRING,
  `ossObjectSize` BIGINT,
  `eventTime` TIMESTAMP(3)
) WITH (
  'connector' = 'print'
);

-- Filter for ObjectCreated events and output results.
INSERT INTO print_sink
SELECT eventName, ossBucketName, ossObjectKey, ossObjectSize, eventTime
FROM oss_event_source
WHERE eventName LIKE 'ObjectCreated:%';