Comment écrire des données de SLS vers Hologres en temps réel - Hologres

Plusieurs méthodes permettent d'écrire des données de Simple Log Service vers Hologres. Cette rubrique explique comment synchroniser les données de Simple Log Service vers Hologres en temps réel à l'aide de Realtime Compute for Apache Flink ou de DataWorks.

Prérequis

Activez le service Log Service, puis créez un projet et un Logstore. Pour plus d'informations, consultez la section Prise en main.
Achetez une instance Hologres et connectez-la à un outil de développement. Pour plus d'informations, consultez la section Procédure d'utilisation de Hologres.
Pour écrire des données de Simple Log Service vers Hologres à l'aide de Realtime Compute for Apache Flink, activez ce service et créez un projet. Pour plus d'informations, consultez les sections Activer Flink entièrement géré et Créer et gérer un namespace.
Pour écrire des données de Log Service vers Hologres à l'aide de DataWorks, activez ce service et créez un espace de travail. Pour plus d'informations, consultez les sections Guide d'achat de DataWorks et Créer un espace de travail.

Contexte

Simple Log Service est une plateforme d'observabilité et d'analyse native du cloud qui fournit des services à grande échelle, économiques et en temps réel pour traiter divers types de données tels que les journaux, les métriques et les traces. Simple Log Service vous permet de collecter, traiter, interroger, analyser, visualiser, consommer et livrer des données. Vous pouvez configurer des alertes dans la console Simple Log Service. Simple Log Service aide les entreprises à améliorer leurs capacités numériques en matière de R&D, d'O&M et de sécurité des données.

Hologres est conçu pour fournir un moteur de calcul en temps réel offrant des performances élevées, une fiabilité accrue, une efficacité économique et une grande évolutivité. Hologres propose des solutions d'entrepôt de données en temps réel qui aident à gérer de grandes quantités de données, ainsi que des services de requête interactive avec des temps de réponse inférieurs à la seconde. Hologres est largement utilisé dans des scénarios tels que la construction de plateformes de données intermédiaires en temps réel, l'analyse granulaire, l'analyse en libre-service, la configuration de profils marketing, le regroupement d'audiences et le contrôle des risques en temps réel. Écrivez rapidement des données de Simple Log Service vers Hologres pour effectuer des analyses et des requêtes en temps réel, ce qui permet d'exploiter pleinement vos données au profit de votre activité.

Écrire des données de Simple Log Service vers Hologres à l'aide de Realtime Compute for Apache Flink

Préparez les données dans Simple Log Service.

Dans cet exemple, les données simulées fournies par Simple Log Service servent de données source. Ces données sont générées en simulant des connexions à une plateforme de jeu et des journaux de consommation. Vous pouvez également utiliser vos propres données métier.
1. Connectez-vous à la console Simple Log Service.
2. Dans la section Import Data, cliquez sur l'onglet Simulated Data Import.
3. Sur l'onglet Simulated Data Import, cliquez sur Simulation dans la section Game Operation Logs.
4. Sur la page Specify Logstore, sélectionnez un project et un Logstore, puis cliquez sur Next.
5. À l'étape Simulated Data Import , configurez les paramètres liés à la plage de temps et à la fréquence, puis cliquez sur Import.
6. La figure suivante présente les champs de simulation et les données que vous pouvez interroger. Pour plus d'informations, consultez la section Interroger et analyser les journaux.
  
  La colonne content est de type JSON. La table de résultats de la requête contient un total de 131 650 enregistrements de journaux de connexion simulés avec les champs suivants : __source__ (valeur : 127.0.0.1), __topic__ (valeur : game_op_log), content (une structure JSON contenant le champ auth), operation (valeur : login), timestamp et uid (un ID utilisateur au format UUID).
Créez une table dans Hologres.
Créez une table dans Hologres destinée à recevoir les données. Créez des index pour les champs de la table en fonction de vos besoins en matière de requêtes afin d'améliorer l'efficacité des interrogations. Pour plus d'informations sur les index, consultez la section Vue d'ensemble. L'instruction DDL suivante permet de créer la table :
```
CREATE TABLE sls_flink_holo (
    content JSONB ,
    operation TEXT,
    uid TEXT,
    topic  TEXT ,
    source TEXT ,
    c__timestamp TIMESTAMPTZ,
    receive_time BIGINT,
    PRIMARY KEY (uid)
  );
```

Utilisez Realtime Compute for Apache Flink pour écrire des données dans Hologres.

Suivez les étapes ci-dessous pour écrire des données de Simple Log Service vers Hologres à l'aide de Realtime Compute for Apache Flink :

Lisez les données de Simple Log Service à l'aide de Realtime Compute for Apache Flink. Pour plus d'informations, consultez la section Créer une table source Simple Log Service.
Écrivez les données dans Hologres à l'aide de Realtime Compute for Apache Flink. Pour plus d'informations, consultez la section Créer une table de résultats Hologres.

L'exemple de job SQL suivant illustre l'écriture de données de Simple Log Service vers Hologres à l'aide de Realtime Compute for Apache Flink. Les champs au format JSON sont écrits directement sous le type de données JSON dans la table Hologres. Le type de données JSON correspond au type VARCHAR dans Realtime Compute for Apache Flink.

Remarque

Pour plus d'informations sur la création et l'exécution de jobs SQL dans Realtime Compute for Apache Flink, consultez les sections Développer un brouillon SQL et Démarrer un déploiement.
Si les données lues depuis Simple Log Service contiennent des données au format JSON, analysez-les avant de les écrire dans Realtime Compute for Apache Flink. Vous pouvez également écrire directement les données au format JSON de Simple Log Service vers Hologres.

CREATE TEMPORARY TABLE sls_input (
    content STRING,
    operation STRING,
    uid STRING,
    `__topic__` STRING METADATA VIRTUAL,
    `__source__` STRING METADATA VIRTUAL,
    `__timestamp__` BIGINT METADATA VIRTUAL,
    `__tag__` MAP<VARCHAR, VARCHAR> METADATA VIRTUAL
  )
WITH (
    'connector' = 'sls',
    'endpoint' = 'Internal endpoint of the Simple Log Service project',-- The internal endpoint that is used to access the Simple Log Service project.
    'accessid'='AccessKey ID of your Alibaba Cloud account',-- The AccessKey ID of your Alibaba Cloud account.
    'accesskey'='AccessKey secret of your Alibaba Cloud account',-- The AccessKey secret of your Alibaba Cloud account.
    'starttime' = '2024-08-30 00:00:00',-- The time at which logs start to be consumed.
    'project'='Project name',-- The name of the Simple Log Service project.
    'logstore'='Logstore name'-- The name of the Logstore. 
  );

CREATE TEMPORARY TABLE hologres_sink (
    content VARCHAR,
    operation VARCHAR,
    uid VARCHAR,
    topic  STRING ,
    source STRING ,
    c__timestamp TIMESTAMP ,
    receive_time BIGINT
  )
WITH (
    'connector' = 'hologres',
    'dbname'='Name of the Hologres database',-- The name of the Hologres database. 
    'tablename'='Name of the Hologres table',-- The name of the Hologres table to which you write data. 
    'username'='AccessKey ID of your Alibaba Cloud account',-- The AccessKey ID of your Alibaba Cloud account. 
    'password'='AccessKey secret of your Alibaba Cloud account',-- The AccessKey secret of your Alibaba Cloud account 
    'endpoint'='VPC endpoint of your Hologres instance.'-- The VPC endpoint of your Hologres instance. 
  );

INSERT INTO hologres_sink
SELECT
   content,
   operation,
   uid,
   `__topic__` ,
   `__source__` ,
    CAST (
    FROM_UNIXTIME (`__timestamp__`) AS TIMESTAMP
  ),
   CAST (__tag__['__receive_time__'] AS BIGINT) AS receive_time
FROM
  sls_input;

Interrogez les données dans Hologres.

Interrogez les données écrites depuis SLS vers Hologres à l'aide de Flink. Procédez ensuite au développement des données en fonction de vos besoins métier. Exécutez l'instruction SELECT * FROM sls_flink_holo;. Les résultats de la requête incluent des champs tels que content, operation, uid, topic, source et c_timestamp. Ces résultats confirment que les données de journaux d'opérations de jeu provenant de SLS, telles que les enregistrements de dépôt, de déconnexion et de connexion, ont été correctement écrites dans Hologres par Flink.

Écrire des données dans Hologres à l'aide de DataWorks

Préparez les données dans Simple Log Service.

Dans cet exemple, les données simulées fournies par Simple Log Service servent de données source. Ces données sont générées en simulant des connexions à une plateforme de jeu et des journaux de consommation. Vous pouvez également utiliser vos propres données métier.
1. Connectez-vous à la console Simple Log Service.
2. Dans la section Import Data, cliquez sur l'onglet Simulated Data Import.
3. Sur l'onglet Simulated Data Import, cliquez sur Simulation dans la section Game Operation Logs.
4. Sur la page Specify Logstore, sélectionnez un project et un Logstore, puis cliquez sur Next.
5. À l'étape Simulated Data Import , configurez les paramètres liés à la plage de temps et à la fréquence, puis cliquez sur Import.
6. La figure suivante présente les champs de simulation et les données que vous pouvez interroger. Pour plus d'informations, consultez la section Interroger et analyser les journaux.
  
  La colonne content est de type JSON. La table de résultats de la requête contient un total de 131 650 enregistrements de journaux de connexion simulés avec les champs suivants : __source__ (valeur : 127.0.0.1), __topic__ (valeur : game_op_log), content (une structure JSON contenant le champ auth), operation (valeur : login), timestamp et uid (un ID utilisateur au format UUID).
Créez une table dans Hologres.
Créez une table dans Hologres destinée à recevoir les données. Créez des index pour les champs de la table en fonction de vos besoins en matière de requêtes afin d'améliorer l'efficacité des interrogations. Pour plus d'informations sur les index, consultez la section Vue d'ensemble. L'instruction DDL suivante permet de créer la table :
Remarque
- Dans cet exemple, le champ uid est configuré comme clé primaire permettant d'identifier les données de manière unique. Configurez la clé primaire en fonction de vos besoins métier.
- Le champ uid est également configuré comme clé de distribution. Ainsi, les enregistrements ayant la même valeur pour le champ uid sont écrits dans le même shard, ce qui améliore les performances des requêtes.
```
BEGIN;
CREATE  TABLE sls_dw_holo (
    content JSONB ,
    operation TEXT,
    uid TEXT,
    C_Topic  TEXT ,
    C_Source TEXT ,
    timestamp BIGINT,
    PRIMARY KEY (uid)
  );
  CALL set_table_property('sls_dw_holo', 'distribution_key', 'uid');
  CALL set_table_property('sls_dw_holo', 'event_time_column', 'timestamp');
COMMIT;
                        
```
Configurez la source de données.
Avant d'écrire des données dans Hologres à l'aide du service Data Integration de DataWorks, ajoutez la source de données Simple Log Service et la source de données Hologres dans l'espace de travail DataWorks spécifié.
- Configurez la source de données Log Service LogHub. Pour plus d'informations, consultez la section Ajouter une source de données LogHub (SLS).
- Configurez la source de données Hologres. Pour plus d'informations, consultez la section Ajouter une source de données Hologres.
Synchronisez les données de Simple Log Service vers Hologres en temps réel.

Créez un nœud de synchronisation en temps réel et exécutez-le dans Data Integration. Pour plus d'informations, consultez les sections Créer un nœud de synchronisation en temps réel pour synchroniser les données incrémentielles d'une seule table et O&M des nœuds de synchronisation en temps réel.

Dans cet exemple, la tâche de synchronisation en temps réel utilise la source de données Loghub en entrée et la source de données Hologres en sortie. Les mappages de champs suivants sont configurés. Dans la boîte de dialogue Field Mapping, mappez les champs source aux champs de destination : content→content, operation→operation, uid→uid, C_Topic→c_topic, C_Source→c_source et timestamp→timestamp. Ajoutez rapidement des mappages en cliquant sur Same-name Mapping ou Same-line Mapping. Après avoir configuré les mappages, cliquez sur OK.
Interrogez les données dans Hologres.

Une fois la tâche de synchronisation en temps réel démarrée, interrogez les données SLS écrites dans Hologres par DataWorks Data Integration. Dans l'éditeur SQL, saisissez l'instruction de requête SELECT * FROM sls_dw_holo; et cliquez sur Run. Les résultats de la requête affichent les données de journaux d'opérations de jeu provenant de SLS écrites dans la table, y compris les champs content, operation, uid, c_topic, c_source et timestamp.