Comment synchroniser les données Doris et champs pris en charge - DataWorks

Versions de Doris prises en charge

Doris Writer utilise le pilote MySQL 5.1.47. Les versions du noyau prises en charge par ce pilote sont indiquées ci-dessous. Pour plus d'informations sur les capacités du pilote, consultez la documentation officielle de Doris.

Version de Doris

Prise en charge

0.x.x

Oui

1,1.x

Oui

1,2.x

Oui

2.x

Oui

Restrictions d'utilisation

Data Integration prend en charge la synchronisation hors ligne des données Doris (lecture et écriture hors ligne).

Types de données pris en charge

Les types de données et les modèles d'agrégation pris en charge varient selon les versions de Doris. Pour obtenir la liste complète des types de données pris en charge par chaque version de Doris, consultez la documentation officielle de Doris. Le tableau suivant présente les principaux types de données pris en charge.

Type de données

Modèle pris en charge

Version de Doris

Écriture hors ligne (Doris Writer)

SMALLINT

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

INT

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

BIGINT

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

LARGEINT

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

FLOAT

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

DOUBLE

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

DECIMAL

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

DECIMALV3

Aggregate, Unique, Duplicate

Versions supérieures à 1.2.1, 2.x

Oui

DATE

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

DATETIME

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

DATEV2

Aggregate, Unique, Duplicate

1,2.x, 2.x

Oui

DATETIMEV2

Aggregate, Unique, Duplicate

1,2.x, 2.x

Oui

CHAR

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

VARCHAR

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

STRING

Aggregate, Unique, Duplicate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

ARRAY

Duplicate

1,2.x, 2.x

Oui

JSON

Aggregate, Unique, Duplicate

1,2.x, 2.x

Oui

HLL

Aggregate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

BITMAP

Aggregate

0.x.x, 1,1.x, 1,2.x, 2.x

Oui

QUANTILE_STATE

Aggregate

1,2.x, 2.x

Oui

Fonctionnement

Doris Writer écrit les données via StreamLoad, une fonctionnalité native de Doris. Doris Writer met en cache les données lues par le lecteur en mémoire, les concatène sous forme de texte, puis importe ce texte par lots dans la base de données Doris. Pour plus d'informations, consultez la documentation officielle de Doris.

Préparer un environnement Doris avant la synchronisation des données

Avant d'utiliser DataWorks pour synchroniser des données vers une source de données Doris, préparez un environnement Doris. Cela garantit que vous pouvez configurer une tâche de synchronisation de données et qu'elle synchronise correctement les données vers la source Doris. Les informations suivantes décrivent comment préparer un environnement Doris pour la synchronisation des données.

Vérifier la version de Doris

Data Integration impose des exigences concernant les versions de Doris. Vérifiez si la version du cluster Doris source répond aux exigences en consultant la section Versions de Doris prises en charge. Téléchargez la version spécifique depuis le site officiel de Doris et installez-la.

Créer un compte et accorder des autorisations

Créez un compte pour vous connecter à la base de données Doris lors des opérations ultérieures. Définissez un mot de passe pour ce compte afin de permettre les connexions futures à la base de données Doris. Si vous souhaitez utiliser l'utilisateur root par défaut de Doris, définissez un mot de passe pour cet utilisateur. Par défaut, l'utilisateur root n'a pas de mot de passe. Exécutez l'instruction SQL suivante dans Doris pour définir le mot de passe :

SET PASSWORD FOR 'root' = PASSWORD('Password')

Configurer la connexion réseau pour Doris

Pour écrire des données via la méthode StreamLoad, accédez à l'adresse IP privée d'un nœud FE. Si vous accédez à l'adresse IP publique du nœud FE, vous êtes redirigé vers l'adresse IP privée d'un nœud BE. Pour plus d'informations sur cette redirection, consultez la page Problèmes liés aux opérations sur les données. Établissez des connexions réseau entre votre source de données et un groupe de ressources serverless ou un groupe de ressources exclusif pour Data Integration, afin de permettre au groupe de ressources d'accéder à la source de données via le réseau interne. Pour savoir comment établir une connexion réseau entre la base de données Doris et un groupe de ressources, consultez les Solutions de connectivité réseau.

Ajouter une source de données

Avant de développer une tâche de synchronisation dans DataWorks, ajoutez la source de données requise à DataWorks en suivant les instructions de la rubrique Configuration de la source de données. Consultez les descriptions des paramètres dans la console DataWorks pour comprendre la signification des paramètres lors de l'ajout d'une source de données.

Tenez compte des exigences de configuration pour les éléments suivants de la source de données Doris :

JdbcUrl : saisissez la chaîne de connexion JDBC, qui inclut l'adresse IP, le numéro de port, la base de données et les paramètres de connexion. Les adresses IP publiques et privées sont prises en charge. Si vous utilisez une adresse IP publique, assurez-vous que le groupe de ressources Data Integration peut accéder à l'hôte où se trouve votre instance Doris.
Point de terminaison FE : saisissez les adresses IP et les ports des nœuds FE. Si votre cluster comporte plusieurs nœuds FE, saisissez plusieurs points de terminaison séparés par des virgules, par exemple ip1:port1,ip2:port2. Lors du test de la connexion, DataWorks teste la connectivité vers tous les points de terminaison FE spécifiés.
Username : saisissez le nom d'utilisateur pour accéder à la base de données Doris.
Password : saisissez le mot de passe correspondant au nom d'utilisateur.
Paramètres avancés : la source de données Doris prend en charge plusieurs paramètres avancés qui étendent la configuration de la connexion. Par exemple, pour ajuster le délai d'expiration des requêtes Java Database Connectivity (JDBC), ajoutez le paramètre queryTimeout aux paramètres avancés. La valeur est exprimée en secondes, et queryTimeout=600 définit le délai d'expiration à 600 secondes. Après avoir défini ce paramètre, il est automatiquement ajouté à la chaîne de connexion JDBC. Le paramètre queryTimeout s'applique uniquement au niveau de la source de données. Vous ne pouvez pas configurer queryTimeout séparément pour un nœud de synchronisation par lots. Configurez-le de manière centralisée dans la source de données.

Développer une tâche de synchronisation de données

Pour connaître le point d'entrée et la procédure de configuration d'une tâche de synchronisation, consultez les guides de configuration suivants.

Pour plus d'informations sur la procédure de configuration, consultez Configurer une tâche de synchronisation par lots à l'aide de l'interface sans code et Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code.
Pour obtenir des informations sur tous les paramètres configurés et le code exécuté lors de l'utilisation de l'éditeur de code pour configurer une tâche de synchronisation par lots, consultez l'Annexe : Code et paramètres.

Annexe : Code et paramètres

Configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code

Si vous souhaitez configurer une tâche de synchronisation par lots à l'aide de l'éditeur de code, configurez les paramètres associés dans le script conformément aux exigences de format de script unifié. Pour plus d'informations, consultez la rubrique Configuration en mode script. Les informations suivantes décrivent les paramètres que vous devez configurer pour les sources de données lors de la configuration d'une tâche de synchronisation par lots à l'aide de l'éditeur de code.

Exemple de script Reader

{
  "type": "job",
  "version": "2.0",// The version number. 
  "steps": [
    {
      "stepType": "doris",// The plug-in name. 
      "parameter": {
        "column": [// The names of the columns. 
          "id"
        ],
        "connection": [
          {
            "querySql": [
              "select a,b from join1 c join join2 d on c.id = d.id;"
            ],
            "datasource": ""// The name of the data source. 
          }
        ],
        "where": "",// The WHERE clause. 
        "splitPk": "",// The shard key. 
        "encoding": "UTF-8"// The encoding format. 
      },
      "name": "Reader",
      "category": "reader"
    },
    {
      "stepType": "stream",
      "parameter": {},
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "errorLimit": {
      "record": "0"// The maximum number of dirty data records allowed. 
    },
    "speed": {
      "throttle": true,// Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled. The mbps parameter takes effect only when the throttle parameter is set to true. 
      "concurrent": 1,// The maximum number of parallel threads. 
      "mbps": "12"// The maximum transmission rate. Unit: MB/s. 
    }
  },
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  }
}

Paramètres du script Reader

Paramètre	Description	Obligatoire	Valeur par défaut
datasource	Le nom de la source de données. Il doit correspondre au nom de la source de données ajoutée. Vous pouvez ajouter des sources de données à l'aide de l'éditeur de code.	Oui	Aucune valeur par défaut
table	Le nom de la table à partir de laquelle vous souhaitez lire les données. Chaque tâche de synchronisation permet de synchroniser les données d'une seule table. Pour une table fragmentée, utilisez le paramètre table pour spécifier les partitions à partir desquelles vous souhaitez lire les données. Exemples : Vous pouvez configurer une plage pour lire les données à partir de bases de données et de tables fragmentées. Par exemple, `'table_[0-99]'` indique que les données sont lues à partir de `'table_0'`, `'table_1'` et `'table_2'`, jusqu'à `'table_99'`. Si vos tables ont des suffixes numériques de même longueur, tels que `'table_000'`, `'table_001'` et `'table_002'`, jusqu'à `'table_999'`, configurez le paramètre comme suit : `'"table": ["table_00[0-9]", "table_0[10-99]", "table_[100-999]"]'`. Remarque Doris Reader lit les données à partir des colonnes spécifiées par le paramètre column dans les partitions spécifiées par le paramètre table. Si une partition ou une colonne spécifiée n'existe pas, la tâche de synchronisation échoue.	Oui	Aucune valeur par défaut
column	Les colonnes que vous souhaitez synchroniser. Les colonnes sont décrites dans un tableau JSON. Par défaut, toutes les colonnes sont synchronisées. Par exemple, utilisez `[""]`. Élagage des colonnes : sélectionnez un sous-ensemble de colonnes à exporter. L'ordre des colonnes peut être modifié. Cela signifie que vous pouvez spécifier les colonnes dans un ordre différent de celui défini par le schéma de la table source. Configuration des constantes : respectez la syntaxe SQL de Doris. Exemple : `["id","table","1","'mingya.wmy'","'null'","to_char(a+1)","2.3","true"]`. id : un nom de colonne. table* : le nom d'une colonne contenant des mots-clés réservés. 1 : une constante entière. 'mingya.wmy' : une constante de chaîne, entourée de guillemets simples ('). null : `" "` indique une chaîne vide. `null` indique une valeur nulle. `'null'` indique la chaîne « null ». to_char(a+1) : une expression de fonction utilisée pour calculer la longueur d'une chaîne. 2,3 : une constante à virgule flottante. true : une valeur booléenne. Le paramètre column doit spécifier explicitement toutes les colonnes à partir desquelles vous souhaitez lire les données. Ce paramètre ne peut pas être laissé vide.	Oui	Aucune valeur par défaut
splitPk	Pour améliorer les performances de lecture, utilisez le paramètre splitPk pour spécifier une clé de fragmentation. Data Integration utilise cette clé pour partitionner les données et exécuter des tâches concurrentes. Nous vous recommandons de définir le paramètre splitPk sur le nom de la colonne de clé primaire de la table. Les données peuvent être distribuées uniformément sur différents fragments en fonction de la colonne de clé primaire, au lieu d'être concentrées uniquement sur des fragments spécifiques. Le paramètre splitPk prend en charge la fragmentation des données uniquement pour les types de données entiers. Si vous définissez le paramètre splitPk sur un champ d'un type de données non pris en charge, tel qu'une chaîne, un nombre à virgule flottante ou une date, le paramètre est ignoré et un seul thread est utilisé pour lire les données. Si le paramètre splitPk n'est pas fourni ou est laissé vide, un seul thread est utilisé pour lire les données.	Non	Aucune valeur par défaut
where	La condition de filtrage. Dans de nombreux scénarios métier, vous souhaiterez peut-être synchroniser uniquement les données du jour en cours. Pour ce faire, spécifiez la condition where comme suit : `gmt_create > $bizdate`. Utilisez la clause WHERE pour lire des données incrémentielles. Si le paramètre where n'est pas fourni ou est laissé vide, Doris Reader lit toutes les données. Ne définissez pas le paramètre where sur limit 10. Cette valeur n'est pas conforme aux contraintes de Doris sur la clause SQL WHERE.	Non	Aucune valeur par défaut
querySql (paramètre avancé, disponible uniquement dans l'éditeur de code)	Dans certains scénarios métier, le paramètre where peut ne pas suffire à décrire les conditions de filtrage souhaitées. Utilisez ce paramètre pour spécifier une requête SQL personnalisée pour le filtrage. Si vous configurez ce paramètre, Data Integration ignore les paramètres table, column, where et splitPk, et utilise la requête personnalisée pour récupérer les données. Par exemple, pour joindre plusieurs tables avant la synchronisation, utilisez une requête telle que `select a,b from table_a join table_b on table_a.id = table_b.id`. Le paramètre querySql a une priorité plus élevée que les paramètres table, column, where et splitPk. Le paramètre datasource reste requis pour que le système récupère les détails de la connexion, tels que le nom d'utilisateur et le mot de passe. Remarque Le nom du paramètre querySql respecte la casse. Par exemple, querysql ne prend pas effet.	Non	Aucune valeur par défaut

Exemple de script Writer

{
  "stepType": "doris",// The plug-in name. 
  "parameter":
  {
    "postSql":// The SQL statement that you want to execute after the synchronization task is run. 
    [],
    "preSql":
    [],// The SQL statement that you want to execute before the synchronization task is run. 
    "datasource":"doris_datasource",// The name of the data source. 
    "table": "doris_table_name",// The name of the table. 
    "column":
    [
      "id",
      "table_id",
      "table_no",
      "table_name",
      "table_status"
    ],
    "loadProps":{
      "column_separator": "\\x01",// The column delimiter of data in the CSV format.
      "line_delimiter": "\\x02"// The row delimiter of data in the CSV format.
    }
  },
  "name": "Writer",
  "category": "writer"
}

Paramètres du script Writer

Paramètre	Description	Obligatoire	Valeur par défaut
datasource	Le nom de la source de données. Il doit correspondre au nom de la source de données ajoutée. Vous pouvez ajouter des sources de données à l'aide de l'éditeur de code.	Oui	Aucune valeur par défaut
table	Le nom de la table à partir de laquelle vous souhaitez lire les données.	Oui	Aucune valeur par défaut
column	Les colonnes de destination vers lesquelles écrire les données. Spécifiez les noms de colonnes dans un tableau, par exemple `"column":["id","name","age"]`. Pour écrire les données dans toutes les colonnes dans l'ordre, utilisez un astérisque () dans le tableau, par exemple `"column":[""]`.	Oui	Aucune valeur par défaut
preSql	Les instructions SQL à exécuter avant le démarrage de la tâche de synchronisation des données. Dans l'interface sans code, vous ne pouvez exécuter qu'une seule instruction SQL. Dans l'éditeur de code, vous pouvez exécuter plusieurs instructions SQL. Par exemple, utilisez ces instructions pour effacer les données existantes de la table.	Non	Aucune valeur par défaut
postSql	L'instruction SQL que vous souhaitez exécuter après l'exécution de la tâche de synchronisation. Par exemple, définissez ce paramètre sur l'instruction SQL utilisée pour ajouter un horodatage. Vous ne pouvez exécuter qu'une seule instruction SQL dans l'interface sans code et plusieurs instructions SQL dans l'éditeur de code.	Non	Aucune valeur par défaut
maxBatchRows	Le nombre maximal de lignes que vous pouvez écrire dans la table de destination en une seule fois. Ce paramètre et le paramètre batchSize déterminent conjointement le nombre d'enregistrements de données que vous pouvez écrire dans la table de destination en une seule fois. Chaque fois que les données mises en cache atteignent la valeur de l'un ou l'autre paramètre, le writer commence à écrire les données dans la table de destination.	Non	500000
batchSize	La quantité maximale de données que vous pouvez écrire dans la table de destination en une seule fois. Ce paramètre et le paramètre maxBatchRows déterminent conjointement le nombre d'enregistrements de données que vous pouvez écrire dans la table de destination en une seule fois. Chaque fois que les données mises en cache atteignent la valeur de l'un ou l'autre paramètre, le writer commence à écrire les données dans la table de destination.	Non	104857600
maxRetries	Le nombre maximal de tentatives autorisées après l'échec de l'écriture de plusieurs enregistrements de données dans la table de destination en une seule fois.	Non	3
labelPrefix	Le préfixe d'étiquette pour chaque lot de fichiers téléchargés. L'étiquette finale est une combinaison de `labelPrefix + UUID` afin de garantir que chaque importation est idempotente, empêchant ainsi la duplication des données.	Non	datax_doris_writer_
loadProps	Les paramètres de requête pour StreamLoad, principalement utilisés pour configurer le format des données d'importation. Par défaut, les données sont importées au format CSV. Si le paramètre loadProps n'est pas configuré, le format CSV par défaut est utilisé, avec `\t` comme délimiteur de colonne et `\n` comme délimiteur de ligne. La configuration par défaut est la suivante : `"loadProps": { "format":"csv", "column_separator": "\t", "line_delimiter": "\n" }` Si vous souhaitez écrire des données au format JSON, utilisez les paramètres suivants : `"loadProps": { "format": "json" }`	Non	Aucune valeur par défaut

Écrire des données de types agrégats

Vous pouvez utiliser Doris Writer pour écrire des données de types agrégats, mais vous devez effectuer des configurations supplémentaires dans l'éditeur de code.

Par exemple, dans la table Doris suivante, uuid est de type bitmap (type d'agrégation) et sex est de type HLL (type d'agrégation).

CREATE TABLE `example_table_1` (
  `user_id` int(11) NOT NULL,
  `date` varchar(10) NOT NULL,
  `city` varchar(10) NOT NULL,
  `uuid` bitmap BITMAP_UNION, -- Aggregation type
  `sex` HLL HLL_UNION  -- Aggregation type
) ENGINE=OLAP AGGREGATE KEY(`user_id`, `date`, `city`)
COMMENT 'OLAP' DISTRIBUTED BY HASH(`user_id`) BUCKETS 32

Les données brutes à synchroniser comportent cinq colonnes, comme indiqué ci-dessous. Les valeurs des colonnes uuid et sex sont des valeurs brutes (nombres ordinaires et chaînes) et non des valeurs de type bitmap ou HLL :

user_id,date,city,uuid,sex
0,T0S4Pb,abc,43,'54'
1,T0S4Pd,fsd,34,'54'
2,T0S4Pb,fa3,53,'64'
4,T0S4Pb,fwe,87,'64'
5,T0S4Pb,gbr,90,'56'
2,iY3GiHkLF,234,100,'54'

Les valeurs brutes ne peuvent pas être écrites directement dans les colonnes bitmap ou HLL. Convertissez les valeurs à l'aide de fonctions d'agrégation. Le paramètre columns de Doris Stream Load peut effectuer cette conversion. La règle de mappage est la suivante : les colonnes du fichier sont mappées séquentiellement aux noms de colonnes qui ne contiennent pas de signe égal (=) dans columns, et les entrées contenant un signe égal (=) sont des transformations d'expression (les valeurs des colonnes de destination sont calculées à partir des expressions) et n'occupent pas de positions de colonne de fichier.

Par conséquent, les mappages entre les colonnes du fichier et les colonnes de la table dans cet exemple sont les suivants :

Colonnes 1 à 3 du fichier (user_id, date et city) : écrites directement dans les colonnes de table portant les mêmes noms.
Colonne 4 du fichier (valeur brute de uuid) : assignée à la colonne temporaire de remplacement k1, convertie en bitmap par l'expression uuid=bitmap_hash(k1), et écrite dans la colonne de table uuid.
Colonne 5 du fichier (valeur brute de sex) : assignée à la colonne temporaire de remplacement k2, convertie en valeur HLL par l'expression sex=hll_hash(k2), et écrite dans la colonne de table sex.

k1 et k2 sont des noms temporaires de remplacement qui ne sont pas écrits dans la table. Vous pouvez modifier ces noms comme bon vous semble, à condition que les noms référencés dans les expressions soient cohérents avec les noms des colonnes de remplacement.

Lorsque vous utilisez Doris Writer pour écrire des données de types agrégats, spécifiez les colonnes à écrire dans writer.parameter.column et configurez les règles de conversion précédentes dans writer.parameter.loadProps.columns. uuid et sex dans writer.parameter.column sont les noms des champs de valeur brute dans les enregistrements de synchronisation. Ils déterminent uniquement les champs à écrire et l'ordre d'écriture. Un fichier CSV ne contient pas de noms de colonnes, et Stream Load identifie les colonnes par leur position. Par conséquent, vous pouvez utiliser des noms personnalisés pour les quatrième et cinquième colonnes dans le paramètre columns (k1 et k2 dans cet exemple).

Exemple de code :

{
    "stepType": "doris",
    "parameter": {
        "datasource": "doris_datasource",
        "table": "example_table_1",
        "column": [
            "user_id",
            "date",
            "city",
            "uuid",
            "sex"
        ],
        "loadProps": {
            "format": "csv",
            "column_separator": "\\x01",
            "line_delimiter": "\\x02",
            "columns": "user_id,date,city,k1,uuid=bitmap_hash(k1),k2,sex=hll_hash(k2)"
        },
        "preSql": [],
        "postSql": []
    },
    "name": "Writer",
    "category": "writer"
}