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

La source de données Doris permet de lire et d'écrire des données dans des bases de données Doris pour le traitement de données à grande échelle. Cette rubrique explique comment utiliser DataWorks pour synchroniser des données avec Doris.

Types de données pris en charge

Les versions de Doris prennent en charge différents types de données et modèles d'agrégation. Pour la liste complète des types de données pris en charge par chaque version, 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

SMALLINT

Aggregate, Unique, Duplicate

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

INT

Aggregate, Unique, Duplicate

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

BIGINT

Aggregate, Unique, Duplicate

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

LARGEINT

Aggregate, Unique, Duplicate

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

FLOAT

Aggregate, Unique, Duplicate

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

DOUBLE

Aggregate, Unique, Duplicate

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

DECIMAL

Aggregate, Unique, Duplicate

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

DECIMALV3

Aggregate, Unique, Duplicate

Versions ultérieures à 1.2.1, 2.x

DATE

Aggregate, Unique, Duplicate

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

DATETIME

Aggregate, Unique, Duplicate

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

DATEV2

Aggregate, Unique, Duplicate

1,2.x, 2.x

DATATIMEV2

Aggregate, Unique, Duplicate

1,2.x, 2.x

CHAR

Aggregate, Unique, Duplicate

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

VARCHAR

Aggregate, Unique, Duplicate

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

STRING

Aggregate, Unique, Duplicate

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

VARCHAR

Aggregate, Unique, Duplicate

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

ARRAY

Duplicate

1,2.x, 2.x

JSONB

Aggregate, Unique, Duplicate

1,2.x, 2.x

HLL

Aggregate

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

BITMAP

Aggregate

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

QUANTILE_STATE

Aggregate

1,2.x, 2.x

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 l'environnement Doris. Cette étape garantit que la tâche de synchronisation peut être configurée et exécutée correctement. La procédure ci-dessous détaille la préparation de l'environnement Doris.

Créer un compte et accorder des autorisations

Créez un compte pour vous connecter à la base de données Doris et effectuez les opérations ultérieures. Définissez un mot de passe pour ce compte afin d'établir les connexions suivantes. Si vous souhaitez utiliser l'utilisateur root par défaut de Doris, définissez-lui un mot de passe (par défaut, l'utilisateur root n'en possède pas). 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. L'accès à l'adresse IP publique du nœud FE redirige vers l'adresse IP privée d'un nœud BE. Pour plus d'informations sur cette redirection, consultez Problèmes liés aux opérations de données. Établissez une connexion 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 l'accès à la source de données via le réseau interne. Pour plus d'informations sur l'établissement de la connectivité entre la base de données Doris et un groupe de ressources, consultez 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 en suivant les instructions de Configuration de la source de données. Consultez les descriptions des paramètres dans la console DataWorks pour comprendre leur signification lors de l'ajout d'une source de données.

Respectez les exigences de configuration suivantes pour la source de données Doris :

JdbcUrl : saisissez la chaîne de connexion JDBC, incluant l'adresse IP, le 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 de votre instance Doris.
Point de terminaison FE : saisissez les adresses IP et les ports des nœuds FE. Pour un cluster comportant 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 connexion, DataWorks vérifie 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 pour étendre la configuration de connexion. Par exemple, pour ajuster le délai d'expiration des requêtes JDBC, ajoutez le paramètre queryTimeout aux paramètres avancés. La valeur est exprimée en secondes ; queryTimeout=600 définit le délai à 600 secondes. Ce paramètre 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 individuellement pour un nœud de synchronisation par lots ; configurez-le centralement 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 suivants.

Pour 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 une tâche de synchronisation par lots, consultez 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

Pour configurer une tâche de synchronisation par lots via l'éditeur de code, définissez les paramètres associés dans le script selon le format unifié. Pour plus d'informations, consultez Configuration en mode script. Les sections suivantes décrivent les paramètres à configurer pour les sources de données lors de l'utilisation de l'éditeur de code.

Démonstration du 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 source. Chaque tâche de synchronisation ne peut synchroniser les données que d'une seule table. Pour une table fragmentée, utilisez le paramètre table pour spécifier les partitions à lire. Exemples : Configurez une plage pour lire les données depuis des bases de données et tables fragmentées. Par exemple, `'table_[0-99]'` spécifie la lecture des données depuis `'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 des colonnes spécifiées par le paramètre column dans les partitions définies 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 à synchroniser, décrites dans un tableau JSON. Par défaut, toutes les colonnes sont synchronisées. Exemple : `[""]`. Élagage des colonnes : sélectionnez un sous-ensemble de colonnes à exporter. Modification de l'ordre : spécifiez les colonnes dans un ordre différent de celui du 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 : nom de colonne. table* : nom de colonne contenant des mots clés réservés. 1 : constante entière. 'mingya.wmy' : constante de chaîne, entre guillemets simples ('). null : `" "` indique une chaîne vide. `null` indique une valeur nulle. `'null'` indique la chaîne « null ». to_char(a+1) : expression de fonction utilisée pour calculer la longueur d'une chaîne. 2,3 : constante à virgule flottante. true : valeur booléenne. Le paramètre column doit spécifier explicitement toutes les colonnes à lire. Il ne peut pas rester 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. Définissez le paramètre splitPk sur le nom de la colonne de clé primaire de la table. Cela permet une distribution uniforme des données sur différents fragments, évitant une concentration sur des fragments spécifiques. Le paramètre splitPk prend en charge la fragmentation uniquement pour les types de données entiers. Si vous définissez ce paramètre sur un champ d'un type non pris en charge (chaîne, nombre à virgule flottante ou date), le paramètre est ignoré et un seul thread lit les données. Si le paramètre splitPk n'est pas fourni ou reste vide, un seul thread lit les données.	Non	Aucune valeur par défaut
where	La condition de filtrage. Dans de nombreux scénarios métier, vous souhaiterez synchroniser uniquement les données du jour en cours. 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 reste 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. 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 est sensible à la casse. Par exemple, querysql ne prend pas effet.	Non	Aucune valeur par défaut

Démonstration du 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 cible dans laquelle écrire les données.	Oui	Aucune valeur par défaut
column	Les colonnes de destination dans 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 à 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 à la fois. Ce paramètre et le paramètre batchSize déterminent conjointement le nombre d'enregistrements de données écrits à la fois. Dès 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 à la fois. Ce paramètre et le paramètre maxBatchRows déterminent conjointement le nombre d'enregistrements de données écrits à la fois. Dès 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 à la 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 l'idempotence de chaque importation, 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

Doris Writer permet d'écrire des données dans des colonnes de types d'agrégation spécifiques. Lors de l'écriture dans ces colonnes, configurez les paramètres supplémentaires requis.

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) NULL,
  `date` varchar(10) NULL DEFAULT "10.5",
  `city` varchar(10) NULL,
  `uuid` bitmap BITMAP_UNION NULL, -- 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

Insérez des données brutes dans la table :

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'

Lorsque vous utilisez Doris Writer pour écrire des données dans une colonne de type agrégat, spécifiez la colonne dans writer.parameter.column et configurez une fonction d'agrégation dans writer.parameter.loadProps.columns. Par exemple, utilisez la fonction d'agrégation bitmap_hash pour la colonne uuid et la fonction d'agrégation hll_hash pour la colonne sex.

Exemple de code :

{
    "stepType": "doris",// The plug-in name. 
    "writer":
    {
        "parameter":
        {
            "column":
            [
                "user_id",
                "date",
                "city",
                "uuid",// The aggregation type is bitmap.
                "sex"// The aggregation type is HLL.
            ],
            "loadProps":
            {
                "format": "csv",
                "column_separator": "\\x01",
                "line_delimiter": "\\x02",
                "columns": "user_id,date,city,k1,uuid=bitmap_hash(k1),k2,sex=hll_hash(k2)"// You must specify the aggregate functions.
            },
            "postSql":
            [
                "select count(1) from example_tbl_3"
            ],
            "preSql":
            [],
            "datasource":"doris_datasource",// The name of the data source. 
                    "table": "doris_table_name",// The name of the table. 
        }
          "name": "Writer",
              "category": "writer"
    }
}