Apache Doris binlog Load use Methods and Examples

1. Configuration de l'installation Mysql

1. InstallationMysql

   Utilisation rapideDockerConfiguration de l'installationMysql,Voir la connexion ci - dessous pour plus de détails

   https://segmentfault.com/a/1190000021523570

   Si vous installez sur une machine physique, reportez - vous aux connexions suivantes：

   In CentOS 7 Installation moyenne MySQL 8 Détails du tutoriel pour

2. OuvertMysql binlog

   Entrée Docker Modification du conteneur ou de la machine physique/etc/my.cnf Documentation,In [mysqld] Ajouter ce qui suit,

   log_bin=mysql_binbinlog-format=Rowserver-id=1

   Puis redémarrezMysql

   systemctl restart mysqld

3. Création Mysql Tableau

   create database demo;    CREATE TABLE `test_cdc` (  `id` int NOT NULL AUTO_INCREMENT,  `sex` TINYINT(1) DEFAULT NULL,  `name` varchar(20) DEFAULT NULL,  `address` varchar(255) DEFAULT NULL,  PRIMARY KEY (`id`) ) ENGINE=InnoDB

2. Configuration de l'installationCanal

Téléchargercanal-1.1.5: https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz

1. DécompresserCanalVers le répertoire spécifié：

   tar zxvf canal.deployer-1.1.5.tar.gz -C ./canal

2. InconfCréer un nouveau répertoire sous le dossier et renommer,En tant queinstanceRacine de,Nom du Répertoire vous pouvez le nommer vous - même pour faciliter l'identification

   Par exemple, mon nom ici correspond au nom de ma base de données：demo

   vi conf/demo/instance.properties

   Voici un exemple de configuration:

   Veuillez vous référer àCanalDocuments officiels：QuickStart

   ################################################### mysql serverId , v1.0.26+ will autoGencanal.instance.mysql.slaveId=12115   # enable gtid use true/falsecanal.instance.gtidon=false   # position infocanal.instance.master.address=10.220.146.11:3306canal.instance.master.journal.name=canal.instance.master.position=canal.instance.master.timestamp=canal.instance.master.gtid=   # rds oss binlogcanal.instance.rds.accesskey=canal.instance.rds.secretkey=canal.instance.rds.instanceId=   # table meta tsdb infocanal.instance.tsdb.enable=true#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb#canal.instance.tsdb.dbUsername=canal#canal.instance.tsdb.dbPassword=canal   #canal.instance.standby.address =#canal.instance.standby.journal.name =#canal.instance.standby.position =#canal.instance.standby.timestamp =#canal.instance.standby.gtid=   # username/passwordcanal.instance.dbUsername=zhangfengcanal.instance.dbPassword=zhangfeng800729)(*Qcanal.instance.connectionCharset = UTF-8# enable druid Decrypt database passwordcanal.instance.enableDruid=false#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==   # table regexcanal.instance.filter.regex=demo\\..*# table black regexcanal.instance.filter.black.regex=# table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)#canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch# table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)#canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch   # mq config#canal.mq.topic=# dynamic topic route by schema or table regex#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*#canal.mq.partition=0# hash partition config#canal.mq.partitionsNum=3#canal.mq.partitionHash=test.table:id^name,.*\\..*#################################################

3. DémarrageCanal

   sh bin/startup.sh

Attention!：canal instance user/passwd

1.1.5 Version,Incanal.propertiesPlus ces deux configurations

canal.user = canal canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458

Le mot de passe par défaut estcanal/canal,canal.passwdLa valeur du mot de passe pourselect password(“xxx”) Pour obtenir

1. Vérifier le démarrage réussi

   tail -200f logs/demo/demo.log

   

3.Début de la synchronisation des données

3.1 CréationDorisTableau des objectifs

L'utilisateur doit d'abordDorisFin créée avecMysqlTableau cible correspondant à l'extrémité

Binlog LoadNe peut supporter queUniqueTableau cible pour le type,Et le tableau cible doit être activéBatch DeleteFonction.

OuvertBatch DeleteLa méthode peut être référencéehelp alter tableFonction de suppression par lots dans.

CREATE TABLE `doris_mysql_binlog_demo` (  `id` int NOT NULL,  `sex` TINYINT(1),  `name` varchar(20),  `address` varchar(255) ) ENGINE=OLAPUNIQUE KEY(`id`,sex)COMMENT "OLAP"DISTRIBUTED BY HASH(`sex`) BUCKETS 1PROPERTIES ("replication_allocation" = "tag.location.default: 3","in_memory" = "false","storage_format" = "V2");-- enable batch deleteALTER TABLE test_2.doris_mysql_binlog_demo ENABLE FEATURE "BATCH_DELETE";

3.1 Créer une tâche de synchronisation

3.1.1 Create Sync Job Description grammaticale

Name: ‘CREATE SYNC JOB’ Description:

Synchronisation des données(Sync Job)Fonction,Permet aux utilisateurs de soumettre une tâche de synchronisation des données résidente,En lisant à partir de l'adresse distante spécifiéeBinlogLog,L'utilisateur de synchronisation incrémentale estMysqlOpération de mise à jour des données de la base de donnéesCDC(Change Data Capture)Fonction.

Actuellement, les tâches de synchronisation des données ne prennent en charge que l'amarrageCanal,DeCanal ServerObtenir résoluBinlogDonnées,Import toDorisIntérieur.

Les utilisateurs peuvent utiliser SHOW SYNC JOB Voir l'état de la tâche de synchronisation des données.

Syntaxe：

CREATE SYNC [db.]job_name ( 	channel_desc,  	channel_desc 	... )binlog_desc

1. job_name

   Synchroniser le nom de l'emploi,Est l'identification unique du travail dans la base de données actuelle,Même chose.job_nameUn seul emploi pour.

2. channel_desc

   Canal de données sous job,Utilisé pour décriremysqlTableau des sourcesdorisRelation cartographique de la table cible.

   Syntaxe：

    FROM mysql_db.src_tbl INTO des_tbl [partitions] [columns_mapping]

   1. mysql_db.src_tbl

      DésignationmysqlBase de données et table source pour.

   2. des_tbl

      DésignationdorisTableau des cibles à la fin,Soutien uniquementUniqueTableau,Et le tableau doit être ouvertbatch deleteFonction(Voir la méthode d'ouverturehelp alter tableDe’Fonction de suppression par lots’).

   3. partitions

      Spécifiez quels tableaux de destination importer partition Moyenne.Si non spécifié,Est automatiquement importé dans le partition Moyenne.

      Exemple：

       PARTITION(p1, p2, p3)

   4. column_mapping

      DésignationmysqlTableau des sources etdorisRelations cartographiques entre les colonnes du tableau cible.Si non spécifié,FEPar défaut, les colonnes du tableau source et du tableau cible correspondent une par une dans l'ordre.

      Non pris en charge col_name = expr Colonne de représentation formelle pour.

      Exemple：

       Supposons que le tableau cible soit listé comme suit:(k1, k2, v1),	      Modifier la colonnek1Etk2Ordre COLUMNS(k2, k1, v1)	      Ignorer la quatrième colonne des données sources COLUMNS(k2, k1, v1, dummy_column)

3. binlog_desc

   Utilisé pour décrire les sources de données distantes,Actuellement seulement pris en chargecanalUn.

   Syntaxe：

    FROM BINLOG (     "key1" = "value1",      "key2" = "value2" )

   1. Canal Propriétés correspondant à la source de données,Parcanal.Préfixe

      1. canal.server.ip: canal serverAdresse
      2. canal.server.port: canal serverLe port de
      3. canal.destination: instanceIdentification
      4. canal.batchSize: AcquisbatchTaille maximale,Par défaut8192
      5. canal.username: instanceNom d'utilisateur pour
      6. canal.password: instanceMot de passe pour
      7. canal.debug: Facultatif,Set totrueHeure,Oui.batchEt les détails de chaque ligne de données sont imprimés Examples:

4. C'est simple. test_db De test_tbl Créer un nom job1 Tâches de synchronisation des données pour,Connexion localeCanalServeur,Correspondant àMysqlTableau des sources mysql_db1.tbl1.

    CREATE SYNC `test_db`.`job1` (     FROM `mysql_db1`.`tbl1` INTO `test_tbl ` ) FROM BINLOG  (     "type" = "canal",     "canal.server.ip" = "127.0.0.1",     "canal.server.port" = "11111",     "canal.destination" = "example",     "canal.username" = "",     "canal.password" = "" );

5. Pour test_db Plusieurs tables pour créer une table nommée job1 Tâches de synchronisation des données pour,Un par un correspond à plusieurs feuillesMysqlTableau des sources,Et spécifie explicitement la cartographie des colonnes.

    CREATE SYNC `test_db`.`job1`  (     FROM `mysql_db`.`t1` INTO `test1` COLUMNS(k1, k2, v1) PARTITIONS (p1, p2),     FROM `mysql_db`.`t2` INTO `test2` COLUMNS(k3, k4, v2) PARTITION p1 )  FROM BINLOG  (     "type" = "canal",      "canal.server.ip" = "xx.xxx.xxx.xx",      "canal.server.port" = "12111",      "canal.destination" = "example",       "canal.username" = "username",      "canal.password" = "password" );

3.1.2 Début de la synchronisationmysqlDonnées dans le tableauDoris

Attention!：

Avant de créer une tâche de synchronisation,Il faut d'abordfe.confConfiguration intérieureenable_create_sync_job=true,Cette valeur par défaut estfalseNe pas activer,Sinon, la tâche de synchronisation ne peut pas être créée

CREATE SYNC test_2.doris_mysql_binlog_demo_job (	FROM demo.test_cdc INTO doris_mysql_binlog_demo) FROM BINLOG (	"type" = "canal", 	"canal.server.ip" = "10.220.146.10", 	"canal.server.port" = "11111", 	"canal.destination" = "demo",  	"canal.username" = "canal", 	"canal.password" = "canal");

3.1.3 Afficher les tâches de synchronisation

SHOW SYNC JOB from test_2;

3.1.4 Voir les données dans le tableau

select * from doris_mysql_binlog_demo;

3.1.5 Supprimer les données

On est là.Mysql Supprimer les données de la table de données,Et regarde.DorisChangements dans l'horloge

delete from test_cdc where id in (12,13)

On y va.DorisÀ l'intérieur.,id- Oui.12,13Ces deux données ont été supprimées

3.1.6 Synchronisation multi - tables

La synchronisation multi - tables n'a besoin que d'écrire comme suit

CREATE SYNC test_2.doris_mysql_binlog_demo_job (	FROM demo.test_cdc INTO doris_mysql_binlog_demo,    FROM demo.test_cdc_1 INTO doris_mysql_binlog_demo,    FROM demo.test_cdc_2 INTO doris_mysql_binlog_demo,    FROM demo.test_cdc_3 INTO doris_mysql_binlog_demo)