Interroger vos données avec Insight

Insight transforme vos flux de topics en entrepôt de données interrogeable sans écrire de code : vous déclarez des tables, décrivez comment les messages s'y associent, puis interrogez le résultat depuis n'importe quel client SQL. Cette page vous guide dans la création d'une instance, la construction d'un pipeline et la connexion avec un client MySQL.

Fonctionnement d'Insight

Une instance est votre propre base de données isolée, avec des ressources, des utilisateurs et des données dédiés. Un projet peut avoir plusieurs instances.
Les tables suivent un schéma en étoile : les tables de faits contiennent les événements (lectures, alertes…), les tables de dimensions décrivent les entités auxquelles ils font référence (appareils, sites…).
Un pipeline lit vos topics, joint chaque fait à ses dimensions et écrit le résultat dans vos tables en continu. Les pipelines s'exécutent sur Apache Flink et les données sont stockées dans Apache Doris.
Les dimensions sont historisées : Insight conserve chaque version d'une entité et, séparément, sa valeur actuelle.
Vous interrogez l'instance via le protocole MySQL, depuis des outils de BI, Grafana, des notebooks ou la ligne de commande mysql.

Avant de commencer

La CLI fun installée et connectée (voir Authentification).
Insight activé pour votre projet.
Les permissions insight list, show, store, update et delete sur le projet (voir Rôles et permissions).
Des topics transportant vos données au format de message Insight (étape 3).

1. Créer une instance

fun api v2 insight create-insight-instance --project-id <project-id> --body '{"name": "plant-analytics"}'

La réponse contient l'id de l'instance. Sa base de données se nomme instance_<instance-id>_database.

2. Créer les tables

Chaque table requiert au moins une colonne clé DATETIME, les colonnes clés placées avant les colonnes de valeurs, une colonne de partitionnement et au moins une colonne de bucketing. Les types de colonnes pris en charge sont BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, LARGEINT, FLOAT, DOUBLE, DECIMAL, DATE, TIME, DATETIME, CHAR, VARCHAR, STRING et JSON.

Une dimension décrivant vos appareils :

fun api v2 insight create-insight-instance-table --project-id <project-id> --instance-id <instance-id> --body '{
  "name": "devices", "version": 1, "type": "dimension",
  "columns": [
    {"name": "collected_at", "type": "DATETIME", "is_key": true},
    {"name": "serial_number", "type": "VARCHAR(64)", "is_key": true},
    {"name": "site", "type": "STRING"}
  ],
  "partitioning": {"column": "collected_at", "unit": "month"},
  "bucketing": {"columns": ["serial_number"]},
  "indexes": []
}'

Une table de faits pour les lectures :

fun api v2 insight create-insight-instance-table --project-id <project-id> --instance-id <instance-id> --body '{
  "name": "readings", "version": 1, "type": "fact",
  "columns": [
    {"name": "measured_at", "type": "DATETIME", "is_key": true},
    {"name": "serial_number", "type": "VARCHAR(64)", "is_key": true},
    {"name": "site", "type": "STRING"},
    {"name": "temperature", "type": "DOUBLE"}
  ],
  "partitioning": {"column": "measured_at", "unit": "day"},
  "bucketing": {"columns": ["serial_number"]},
  "indexes": []
}'

Dans la base de données, une table de faits se nomme <name>_v<version> (readings_v1). Une dimension crée deux tables : <name>_v<version>_history avec chaque version de chaque entité, et <name>_v<version>_last avec uniquement la plus récente.

Récupérez les identifiants de colonnes nécessaires au pipeline :

fun api v2 insight get-insight-instance-table --project-id <project-id> --instance-id <instance-id> --table-id <table-id>

3. Envoyer des données au format Insight

Les pipelines lisent des messages JSON dont les champs se trouvent dans un objet data. Les noms de champs sont ceux que vous déclarez sur le pipeline :

{"data": {"measured_at": "2026-10-22 14:30:00", "serial_number": "A1B2C3", "temperature": "21.5"}}

Les valeurs sont converties vers le type de la colonne. Les dates-heures utilisent une espace (YYYY-MM-DD HH:MM:SS) ; pour les valeurs avec un décalage UTC, comme les horodatages issus des flux d'événements de la plateforme, appliquez la transformation mv_to_utc au champ.

La télémétrie des appareils arrive dans vos topics encapsulée et encodée en base64 ; une fonction cloud est donc le moyen habituel de la décoder et d'écrire ce format dans un topic dédié.

4. Configurer le pipeline

Écrivez le pipeline dans pipeline.json. Les références de champs (source_field_ref, source_join_field_ref, dimension_join_field_ref, dimension_field_ref) sont des positions à partir de 0 dans les listes fields :

{
  "sources": [
    {"name": "readings", "topic_id": <readings-topic-id>, "debug": false,
     "fields": [{"name": "measured_at"}, {"name": "serial_number"}, {"name": "temperature"}]}
  ],
  "dimensions": [
    {"name": "devices", "topic_id": <devices-topic-id>, "table_id": <devices-table-id>,
     "fields": [
       {"name": "collected_at", "column_id": <collected-at-column-id>},
       {"name": "serial_number", "column_id": <serial-number-column-id>},
       {"name": "site", "column_id": <site-column-id>}
     ]}
  ],
  "sinks": [
    {"name": "readings_sink", "source_name": "readings", "table_id": <readings-table-id>, "weight": 1,
     "mappings": [
       {"column_id": <measured-at-column-id>, "source_field_ref": 0},
       {"column_id": <serial-number-column-id>, "source_field_ref": 1},
       {"column_id": <temperature-column-id>, "source_field_ref": 2},
       {"column_id": <site-column-id>, "dimension": "devices",
        "source_join_field_ref": 1, "dimension_join_field_ref": 1, "dimension_field_ref": 2}
     ]}
  ]
}

Le dernier mapping est la jointure : il recherche l'appareil dont le serial_number correspond à la lecture et écrit son site dans le fait. Un mapping peut aussi écrire une constante avec source_value, et un sink peut filtrer les enregistrements avec une condition. Deux mappings ne peuvent pas cibler la même colonne.

Enregistrez-le, puis soumettez-le :

fun api v2 insight save-insight-instance-pipeline --project-id <project-id> --instance-id <instance-id> --body-file pipeline.json
fun api v2 insight submit-insight-instance-pipeline --project-id <project-id> --instance-id <instance-id>

Vérifiez son statut (starting, running, stopping, stopped, failed…) :

fun api v2 insight get-insight-instance-pipeline --project-id <project-id> --instance-id <instance-id>

Pour l'arrêter, utilisez cancel-insight-instance-pipeline. Une table ne peut pas être supprimée tant qu'un sink l'utilise.

5. Créer un utilisateur de base de données

fun api v2 insight create-insight-user --project-id <project-id> --instance-id <instance-id> \
  --body '{"username": "bi-reader", "password": "<password>"}'

Les noms d'utilisateur comptent de 3 à 32 caractères (lettres, chiffres, ., _, - et @) ; les mots de passe comptent de 12 à 128 caractères. Les utilisateurs de base de données disposent d'un accès en lecture seule à la base de données de l'instance. Listez-les avec list-insight-instance-users et supprimez-en un avec delete-insight-user.

6. Interroger depuis n'importe quel client SQL

Paramètre	Valeur
Hôte	`insight.fundamentum-iot.com`
Port	`9030`
Protocole	MySQL
Base de données	`instance_<instance-id>_database`
Utilisateur et mot de passe	L'utilisateur de base de données de l'étape 5

mysql -h insight.fundamentum-iot.com -P 9030 -u bi-reader -p instance_<instance-id>_database

-- Average temperature per site and day
SELECT site, DATE(measured_at) AS day, AVG(temperature) AS avg_temp
FROM readings_v1
GROUP BY site, DATE(measured_at)
ORDER BY day, site;

-- Where each device is now, and everywhere it has been
SELECT serial_number, site FROM devices_v1_last;
SELECT serial_number, site, collected_at FROM devices_v1_history ORDER BY serial_number, collected_at;

Et ensuite

Référence de l'API dans Swagger UI : Insight
Analyser avec Insight : ce qu'Insight apporte à votre organisation.
Fonctions cloud : mettez en forme les données des appareils pour vos pipelines.
Référence de la Public API : schémas Insight complets (instances, tables, pipeline, utilisateurs).