首页 > 数据库 >logstash实现Mysql(Tidb)数据到Elasticsearch数据迁移(增量同步)

logstash实现Mysql(Tidb)数据到Elasticsearch数据迁移(增量同步)

时间:2024-01-28 21:02:42浏览次数:32  
标签:jdbc last 9200 Mysql 192.168 同步 Elasticsearch Tidb logstash

1、下载Logstash和Elasticsearch,需要版本一致,本次使用版本均为7.8.0版本

[root@zjk data]# ll elasticsearch-7.8.0-linux-x86_64.tar.gz logstash-7.8.0.tar.gz 
-rw-r--r--. 1 root root 319112561 12月 28 15:39 elasticsearch-7.8.0-linux-x86_64.tar.gz
-rw-r--r--. 1 root root 167162975 1月  24 22:13 logstash-7.8.0.tar.gz

2、Elasticsearch的安装与启动此处不赘述,可浏览 elasticsearch安装部署(单机和集群)

3、因为Logstash是用JRuby语言开发的,故需要先装JDK,此次使用的是jdk1.8

4、此次迁移的数据库是Tidb 5.7.25-TiDB-v6.3.0(Mysql),所以驱动使用Mysql驱动,自行下载mysql-connector-java-8.0.27.jar,放至 logstash-7.8.0/logstash-core/lib/jars/

5、编写配置文件(重要),此处以单表同步为例

input {
    jdbc {
        type => "_doc"
         # 数据库连接地址
        jdbc_connection_string => "jdbc:mysql://ip:port/dbname?characterEncoding=UTF-8&autoReconnect=true"
         # 数据库连接账号密码;
        jdbc_user => "root"
        jdbc_password => "root"
         # MySQL依赖包路径;
        jdbc_driver_library => "/data/logstash-7.8.0/logstash-core/lib/jars/mysql-connector-java-8.0.27.jar"
         # the name of the driver class for mysql 低版本mysql 使用 com.mysql.jdbc.Driver
        jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
         # 数据库重连尝试次数
        connection_retry_attempts => "3"
         # 判断数据库连接是否可用,默认false不开启
        jdbc_validate_connection => "true"
         # 数据库连接可用校验超时时间,默认3600S
        jdbc_validation_timeout => "3600"
         # 开启分页查询(默认false不开启);
        jdbc_paging_enabled => "true"
         # 单次分页查询条数(默认100000,若字段较多且更新频率较高,建议调低此值);
        jdbc_page_size => "100000"
         # statement为查询数据sql,如果sql较复杂,建议配通过statement_filepath配置sql文件的存放路径;
         # sql_last_value为内置的变量,存放上次查询结果中最后一条数据tracking_column的值,此处即为ModifyTime;
         # statement_filepath => "mysql/jdbc.sql"
        statement => "select a.* from (select * from tablename order by id) a where a.id > :sql_last_value"
         # 是否将字段名转换为小写,默认true(如果有数据序列化、反序列化需求,建议改为false);
        lowercase_column_names => false
         # Value can be any of: fatal,error,warn,info,debug,默认info;
        sql_log_level => warn
         #
         # 是否记录上次执行结果,true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中;
        record_last_run => true
         # 需要记录查询结果某字段的值时,此字段为true,否则默认tracking_column为timestamp的值;
        use_column_value => true
         # 需要记录的字段,用于增量同步,需是数据库字段
        tracking_column => "id"
         # Value can be any of: numeric,timestamp,Default value is "numeric"
        tracking_column_type => "numeric"
         # record_last_run上次数据存放位置;
        last_run_metadata_path => "/data/logstash-7.8.0/config/track_time"
         # 是否清除last_run_metadata_path的记录,需要增量同步时此字段必须为false;
        clean_run => false
         #
         # 同步频率(分 时 天 月 年),默认每分钟同步一次;
         #
        schedule => "* * * * *"
    }
}
 
output {
    elasticsearch {
         # host => "localhost"
         # port => "9200"
         # 配置ES地址,集群配置方式 ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
        hosts => ["ip:9200"]
         # 索引名字,必须小写,一般以同步的表名为索引名称
        index => "index_name"
         # 数据唯一索引(建议使用数据库KeyID)
        document_id => "%{id}"
    }
    stdout {
        codec => json_lines
    }
}

6、启动logstash

./logstash -f ../config/logstash-mysql2es.conf

 7、如需同时迁移多张表至Elasticsearch,则配置如下

input {
	stdin {}
	jdbc {
		 # 多表同步时,表类型区分,建议命名为“库名_表名”,每个jdbc模块需对应一个type;
		type => "TestDB_DetailTab"
		
		 # 其他配置此处省略,参考单表配置
		 # ...
		 # ...
		 # record_last_run上次数据存放位置;
		last_run_metadata_path => "mysql\last_id.txt"
		 # 是否清除last_run_metadata_path的记录,需要增量同步时此字段必须为false;
		clean_run => false
		 #
		 # 同步频率(分 时 天 月 年),默认每分钟同步一次;
		schedule => "* * * * *"
	}
	jdbc {
		 # 多表同步时,表类型区分,建议命名为“库名_表名”,每个jdbc模块需对应一个type;
		type => "TestDB_Tab2"
		# 多表同步时,last_run_metadata_path配置的路径应不一致,避免有影响;
		 # 其他配置此处省略
		 # ...
		 # ...
	}
}
 
filter {
	json {
		source => "message"
		remove_field => ["message"]
	}
}
 
output {
	# output模块的type需和jdbc模块的type一致
	if [type] == "TestDB_DetailTab" {
		elasticsearch {
			 # host => "192.168.1.1"
			 # port => "9200"
			 # 配置ES集群地址
			hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
			 # 索引名字,必须小写
			index => "detailtab1"
			 # 数据唯一索引(建议使用数据库KeyID)
			document_id => "%{KeyId}"
		}
	}
	if [type] == "TestDB_Tab2" {
		elasticsearch {
			# host => "192.168.1.1"
			# port => "9200"
			# 配置ES集群地址
			hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
			# 索引名字,必须小写
			index => "detailtab2"
			# 数据唯一索引(建议使用数据库KeyID)
			document_id => "%{KeyId}"
		}
	}
	stdout {
		codec => json_lines
	}
}

  

 

logstash 特性:

  • 无需开发,无需开启myslq binlog等,仅需安装配置logstash即可;
  • 凡是SQL可以实现的logstash均可以实现(本就是通过sql查询数据)
  • 支持每次全量同步或按照特定字段(如递增ID、修改时间)增量同步;
  • 同步频率可控,最快同步频率每分钟一次(如果对实效性要求较高,慎用);
  • 不支持被物理删除的数据同步物理删除ES中的数据(可在表设计中增加逻辑删除字段IsDelete标识数据删除)。
  • 如使用递增ID作为增量同步依据,则要考虑分布式数据库中,生成id为分段式生成,id可能会变小。

 



标签:jdbc,last,9200,Mysql,192.168,同步,Elasticsearch,Tidb,logstash
From: https://blog.51cto.com/u_15002821/9454200

相关文章

  • Python elasticsearch-py类库基础用法
    实践环境https://pypi.org/project/elasticsearch/pipinstallelasticsearch==7.6.0离线安装包及依赖包下载地址:https://files.pythonhosted.org/packages/f5/71/45d36a8df68f3ebb098d6861b2c017f3d094538c0fb98fa61d4dc43e69b9/urllib3-1.26.2-py2.py3-none-any.whl#sha256=......
  • k8s之helm部署mysql集群
    一、简介HelmHelm是Kubernetes的包管理器。ChartHelm使用的包格式称为chart。chart存储在ChartRepository。chart就是一个描述Kubernetes相关资源的文件集合。单个chart可以用来部署一些简单的,类似于memcachepod,或者某些复杂的HTTP服务器以及web全栈应用、数据库、缓......
  • 【ElasticSearch】入门-基础概念
    什么是ES?是一个高可用分布式的搜索引擎。可以用于实时存储、检索数据。底层是使用Lucene全文检索框架。基本概念存储结构:由_index_type和_id标识唯一的一个文档_index:指向一个或多个物理分片的逻辑命名空间_type:用于区分同一个集合中的不同的细分(ES6.X中只允许一个i......
  • Mysql数据库更新RedHat/CentOS 从 8.0.14 到 8.0.33,又从8.0.33更新到8.0.35
    sudosystemctlstartmysqldFirstlyweneedbackupalldatabasedataintonewfile,IuseTestPortal.sql/data/VMs_Share/Homes/bell-bash-4.2$mysqldump-uroot-p--databasesTestPortal>TestPortal.sqlEnterpassword:-bash-4.2$2.downloadtheve......
  • MySQL 事务详解
    在数据库中,事务是一组操作单元,它们被视为一个独立的工作单元,要么完全执行,要么完全不执行。MySQL提供了强大的事务支持,允许开发者以一致的方式管理数据库操作。1.事务基础首先,我们需要了解事务的基础概念。一个事务通常包括以下几个关键特性:原子性(Atomicity):事务中的所有操作......
  • GOLAND 中 使用GORM ,出现 未发现“mysql”driver
    错误描述:sql:unknowndriver"mysql"(forgottenimport?)panic:runtimeerror:invalidmemoryaddressornilpointerdereference 解决过程:1、清除缓存2、添加包import_"github.com/go-sql-driver/mysql"_(下划线)用于导入包但不使用它的情况。......
  • MySQL 8.0.26 新增参数 group_replication_view_change_uuid
    MySQL8.0.26新增参数group_replication_view_change_uuidGreatSQL[root@localhost][test]>showglobalvariableslike'group_replication_view_change_uuid';+------------------------------------+-----------+|Variable_name|V......
  • MySQL的CHAR、VARCHAR、TEXT等字符串字段在等值比较("=")时,会忽略掉尾部的空格,导致有空
    一、现象CREATETABLE`user_info`(`id`BIGINT(20)UNSIGNEDNOTNULLAUTO_INCREMENTCOMMENT'主键自增ID',`user_name`VARCHAR(64)NOTNULLDEFAULT''COMMENT'名字',PRIMARYKEY(`id`),KEY`idx_user_name`(`user_name`))E......
  • mysql连接报错:The server time zone value '�й���׼ʱ��'
    java.sql.SQLException:Theservertimezonevalue'�й���׼ʱ��'isunrecognizedorrepresentsmorethanonetimezone.YoumustconfigureeithertheserverorJDBCdriver(viatheserverTimezoneconfigurationproperty)touseamorespecifctimez......
  • MySQL-14.视图
    1.常见的数据库对象对象描述表(TABLE)表是存储数据的逻辑单元,以行和列的形式存在,列就是字段,行就是记录数据字典也叫系统表,存放数据库相关信息的表。系统表的数据通常由数据库系统维护,程序员通常不应该修改,只可查看约束(CONSTRAINT)执行数据校验的规则,用于保证......