DataWorks:Source de données LogHub (SLS)

Simple Log Service (SLS) prend en charge les scénarios de synchronisation de données suivants :

• Synchronisation des données entre LogHub et des sources telles que MaxCompute, sur différentes régions.

• Synchronisation des données entre LogHub et des sources telles que MaxCompute, sur différents comptes Alibaba Cloud.

• Synchronisation des données entre LogHub et des sources telles que MaxCompute, au sein d'un même compte Alibaba Cloud.

• Synchronisation des données entre LogHub et des sources telles que MaxCompute, entre les comptes du cloud public et ceux du China Finance Cloud.

Configurer une source de données

Avant de développer une tâche de synchronisation, vous devez créer la source de données correspondante dans DataWorks. Pour la procédure, consultez Créer une source de données. Pour une description détaillée des paramètres de configuration, reportez-vous aux info-bulles affichées sur la page de configuration.

Créer une source de données entre comptes Alibaba Cloud

L'exemple suivant explique comment utiliser le compte B pour configurer une tâche de synchronisation dans Data Integration afin de transférer les données LogHub du compte A vers MaxCompute du compte B. Les points d'attention spécifiques aux scénarios intercomptes sont les suivants :

1. Créez une source de données LogHub en utilisant l'AccessKey ID et l'AccessKey Secret du compte A.

   Dans ce cas, le compte B peut synchroniser les données de tous les projets SLS appartenant au compte A.

2. Créez une source de données LogHub en utilisant l'AccessKey ID et l'AccessKey Secret de l'utilisateur RAM A1 rattaché au compte A.

   • Le compte A accorde à l'utilisateur RAM A1 les autorisations générales sur SLS, à savoir les stratégies AliyunLogFullAccess et AliyunLogReadOnlyAccess. Pour plus d'informations, consultez Créer un utilisateur RAM et lui attribuer des autorisations.

     Remarque

     Une fois les stratégies système AliyunLogFullAccess et AliyunLogReadOnlyAccess attachées à un utilisateur RAM, celui-ci peut interroger toutes les ressources SLS du compte Alibaba Cloud.

   • Le compte A accorde à l'utilisateur RAM A1 une autorisation personnalisée sur SLS.

     Connectez-vous à la console RAM console > Permission Management > Permission Policy avec le compte A, puis cliquez sur Create Permission Policy.

     Pour en savoir plus sur l'autorisation associée, consultez Présentation du contrôle d'accès et Exemples d'autorisations personnalisées RAM.

     Après avoir autorisé l'utilisateur RAM A1 selon la stratégie suivante, le compte B ne pourra synchroniser les données que depuis les projets SLS project_name1 et project_name2 via cet utilisateur.

     {
         "Version": "1",
         "Statement": [
             {
                 "Action": [
                     "log:Get*",
                     "log:List*",
                     "log:CreateConsumerGroup",
                     "log:UpdateConsumerGroup",
                     "log:DeleteConsumerGroup",
                     "log:ListConsumerGroup",
                     "log:ConsumerGroupUpdateCheckPoint",
                     "log:ConsumerGroupHeartBeat",
                     "log:GetConsumerGroupCheckPoint"
                 ],
                 "Resource": [
                     "acs:log:*:*:project/project_name1",
                     "acs:log:*:*:project/project_name1/*",
                     "acs:log:*:*:project/project_name2",
                     "acs:log:*:*:project/project_name2/*"
                 ],
                 "Effect": "Allow"
             }
         ]
     }

Guide de configuration des tâches de synchronisation hors ligne mono-table

• Pour la procédure, consultez Configurer une tâche de synchronisation hors ligne mono-table dans l'interface sans code et Configurer une tâche de synchronisation hors ligne mono-table à l'aide d'un script.

  Remarque

  Lorsque vous configurez un nœud de synchronisation en mode interface sans code, le format des paramètres doit être cohérent avec celui utilisé en mode script.

• Pour la liste complète des paramètres et un exemple de script en mode script, consultez Paramètres du mode script ci-dessous.

Guide de configuration des tâches de synchronisation en temps réel mono-table

Pour la procédure, consultez Configurer une tâche de synchronisation en temps réel mono-table.

Guide de configuration de la synchronisation en temps réel sur toute la base de données et autres synchronisations au niveau de la base

Pour la procédure, consultez Configurer une tâche de synchronisation en temps réel sur toute la base de données.

Configuration d'une tâche de synchronisation par lots via l'éditeur de code

Pour configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code, vous devez définir les paramètres appropriés dans le script, conformément aux exigences du format de script unifié. Pour plus d'informations, consultez la section Configuration en mode script. Les informations ci-dessous détaillent les paramètres à configurer pour les sources de données lors de la mise en place d'une tâche de synchronisation par lots via l'éditeur de code.

Exemple de script pour le composant Reader

Lecture depuis un Logstore (exemple de base)

{ "type":"job", "version":"2.0",// The version number. "steps":[ { "stepType":"LogHub",// The plug-in name. "parameter":{ "datasource":"",// The data source. "column":[// The columns. "col0", "col1", "col2", "col3", "col4", "__category__", "__source__", "__topic__", // The log topic. "__machineUUID__", // The unique identifier of the collection machine. "__tag__:__hostname__", // The hostname. "__tag__:__path__", // The path. "__time__" // The event time. ], "beginDateTime":"",// The start time for data consumption. "batchSize":"",// The number of data entries queried from SLS at a time. "endDateTime":"",// The end time for data consumption. "fieldDelimiter":",",// The column delimiter. "logstore":""// The name of the destination Logstore. }, "name":"Reader", "category":"reader" }, { "stepType":"stream", "parameter":{}, "name":"Writer", "category":"writer" } ], "setting":{ "errorLimit":{ "record":"0"// The number of error records. }, "speed":{ "throttle":true,// If throttle is set to false, the mbps parameter does not take effect, which means throttling is disabled. If throttle is set to true, throttling is enabled. "concurrent":1 // The number of concurrent threads. "mbps":"12",// The throttling rate. 1 mbps = 1 MB/s. } }, "order":{ "hops":[ { "from":"Reader", "to":"Writer" } ] } }

Lecture depuis un Logstore après traitement des données avec SPL (exemple)

{ "type":"job", "version":"2.0",// The version number. "steps":[ { "stepType":"LogHub",// The plug-in name. "parameter":{ "datasource":"",// The data source. "column":[// The columns. "col100", "col1" ], "beginDateTime":"",// The start time for data consumption. "batchSize":"",// The number of data entries queried from SLS at a time. "endDateTime":"",// The end time for data consumption. "fieldDelimiter":",",// The column delimiter. "logstore":"",// The name of the destination Logstore. "query":"* | where regexp_like(col0, '[0-9]+') | project col100=col0, col1 " // SPL }, "name":"Reader", "category":"reader" }, { "stepType":"stream", "parameter":{}, "name":"Writer", "category":"writer" } ], "setting":{ "errorLimit":{ "record":"0"// The number of error records. }, "speed":{ "throttle":true,// If throttle is set to false, the mbps parameter does not take effect, which means throttling is disabled. If throttle is set to true, throttling is enabled. "concurrent":1 // The number of concurrent threads. "mbps":"12",// The throttling rate. 1 mbps = 1 MB/s. } }, "order":{ "hops":[ { "from":"Reader", "to":"Writer" } ] } }

Lecture depuis un Logstore après extension des champs avec SPL (exemple)

{ "type":"job", "version":"2.0",// The version number. "steps":[ { "stepType":"LogHub",// The plug-in name. "parameter":{ "datasource":"",// The data source. "column":[// The columns. "col0", "col1", "col2", "col3", "col4", "__category__", "__source__", "__topic__", // The log topic. "__machineUUID__", // The unique identifier of the collection machine. "__tag__:__hostname__", // The hostname. "__tag__:__path__", // The path. "__time__", // The event time. "col100", "col101" ], "beginDateTime":"",// The start time for data consumption. "batchSize":"",// The number of data entries queried from SLS at a time. "endDateTime":"",// The end time for data consumption. "fieldDelimiter":",",// The column delimiter. "logstore":"",// The name of the destination Logstore. "query":"* | where regexp_like(col0, '[0-9]+') | extend col100=cast(col2 as BIGINT), extend col101=date_parse(col3, '%Y-%m-%d %H:%i') ",// SPL }, "name":"Reader", "category":"reader" }, { "stepType":"stream", "parameter":{}, "name":"Writer", "category":"writer" } ], "setting":{ "errorLimit":{ "record":"0"// The number of error records. }, "speed":{ "throttle":true,// If throttle is set to false, the mbps parameter does not take effect, which means throttling is disabled. If throttle is set to true, throttling is enabled. "concurrent":1 // The number of concurrent threads. "mbps":"12",// The throttling rate. 1 mbps = 1 MB/s. } }, "order":{ "hops":[ { "from":"Reader", "to":"Writer" } ] } }

Paramètres du script Reader

Si des données sont manquantes lors de la lecture depuis LogHub, vérifiez dans la console LogHub si le champ de métadonnées receive_time se situe bien dans la plage temporelle configurée pour la tâche.

Partitionnement par champs métier personnalisés

Le filtre temporel appliqué aux tâches de synchronisation hors ligne LogHub (SLS) repose toujours sur __tag__:__receive_time__. Le filtrage ou le partitionnement direct par des champs métier personnalisés tels que start_time n'est pas pris en charge. Pour partitionner les données dans MaxCompute selon un champ métier personnalisé, utilisez la solution de contournement suivante :

1. Configurez une tâche de synchronisation hors ligne pour transférer toutes les données LogHub (SLS) vers une table intermédiaire MaxCompute (sans spécifier de champ de partition).

2. Créez un nœud SQL MaxCompute et exécutez l'instruction suivante pour écrire les données dans la table partitionnée de destination en utilisant un champ métier personnalisé (tel que start_time) :

   INSERT OVERWRITE TABLE target_table PARTITION(start_time)
   SELECT col1, col2, start_time
   FROM temp_table;

3. Utilisez des paramètres de planification pour transmettre dynamiquement les valeurs de partition et mettre en œuvre un partitionnement automatique par jour ou par heure.

Exemple de script pour le composant Writer

{ "type": "job", "version": "2.0",// The version number. "steps": [ { "stepType": "stream", "parameter": {}, "name": "Reader", "category": "reader" }, { "stepType": "LogHub",// The plug-in name. "parameter": { "datasource": "",// The data source. "column": [// The columns. "col0", "col1", "col2", "col3", "col4", "col5" ], "topic": "",// The selected topic. "batchSize": "1024",// The number of records to submit at a time. "logstore": ""// The name of the destination LogService Logstore. }, "name": "Writer", "category": "writer" } ], "setting": { "errorLimit": { "record": ""// The number of error records. }, "speed": { "throttle":true,// If throttle is set to false, the mbps parameter does not take effect and bandwidth is not throttled. If throttle is set to true, bandwidth is throttled. "concurrent":3, // The number of concurrent threads. "mbps":"12"// The throttling rate. 1 mbps = 1 MB/s. } }, "order": { "hops": [ { "from": "Reader", "to": "Writer" } ] } }

Paramètres du script Writer

Le composant Writer LogHub (SLS) récupère les données générées par le composant Reader via le framework Data Integration, puis convertit tous les types de données pris en charge en STRING un par un. Lorsque la taille de lot (batchSize) spécifiée est atteinte, les données sont envoyées à LogHub (SLS) en un seul lot à l'aide du SDK Java LogService.