1. 全量配置
products索引字段展示
PUT /products/
{
"mappings": {
"properties": {
"name":{
"type": "text",
"analyzer": "ik_smart"
},
"long_name":{
"type": "text",
"analyzer": "ik_smart"
},
"brand_id":{
"type": "integer"
},
"category_id":{
"type":"integer"
},
"category":{
"type": "keyword"
},
"category_path":{
"type": "keyword"
},
"shop_id":{
"type":"integer"
},
"price":{
"type":"scaled_float",
"scaling_factor":100
},
"sold_count":{
"type":"integer"
},
"review_count":{
"type":"integer"
},
"status":{
"type":"integer"
},
"create_time" : {
"type" : "date"
},
"last_time" : {
"type" : "date"
}
}
}
}
categorys索引字段展示
PUT /categorys/
{
"mappings": {
"properties": {
"name":{
"type": "text",
"analyzer": "ik_smart"
},
"parent_id":{
"type": "integer"
},
"is_directory":{
"type":"integer"
},
"level":{
"type": "integer"
},
"path":{
"type": "text"
},
"create_time" : {
"type" : "date"
},
"last_time" : {
"type" : "date"
},
"delete_time" : {
"type" : "date"
}
}
}
}
配置文件内容展示(同时同步products和categorys)
input {
stdin { }
jdbc {
type => 'products'
#注意mysql连接地址一定要用ip，不能使用localhost等
jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
jdbc_user => "dark"
jdbc_password => "mysql"
#这个jar包的地址是容器内的地址
jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
#每次同步数量
jdbc_page_size => "50000"
statement => "select a.id,a.`name`,a.long_name,a.brand_id,a.three_category_id as category_id,a.shop_id,a.price,a.status,a.sold_count,a.review_count,a.create_time,a.last_time,b.`name` as category,b.path as category_path from lmrs_products as a LEFT JOIN lmrs_product_categorys as b on a.three_category_id = b.id"
schedule => "* * * * *"
}
jdbc {
type => 'categorys'
#注意mysql连接地址一定要用ip，不能使用localhost等
jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
jdbc_user => "dark"
jdbc_password => "mysql"
#这个jar包的地址是容器内的地址
jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => "true"
#每次同步数量
jdbc_page_size => "50000"
statement => "SELECT id,`name`,parent_id,is_directory,`level`,path,create_time,last_time,delete_time FROM lmrs_product_categorys"
schedule => "* * * * *"
}
}
output {
if [type] == "products" {
elasticsearch {
#注意es连接地址一定要用ip，不能使用localhost等
hosts => "172.17.0.7:9200"
index => "products"
document_type => "_doc"
document_id => "%{id}"
}
}
if [type] == "categorys" {
elasticsearch {
#注意es连接地址一定要用ip，不能使用localhost等
hosts => "172.17.0.7:9200"
index => "categorys"
document_type => "_doc"
document_id => "%{id}"
}
}
stdout {
codec => json_lines
}
}
和单表同步的区别就是创建多个jdbc并且在jdbc中添加type，output中对type进行判断实现多表同步。
2. 增量配置
attribute索引字段展示
PUT /attribute
{
"mappings": {
"properties": {
"name":{
"type": "keyword"
},
"value":{
"type":"keyword"
},
"category_id":{
"type": "integer"
},
"attribute_sort":{
"type":"integer"
},
"attribute_value_sort":{
"type": "integer"
},
"category":{
"type": "keyword"
},
"category_path":{
"type":"text"
}
}
}
}
products索引字段展示
PUT /products/
{
"mappings": {
"properties": {
"name":{
"type": "text",
"analyzer": "ik_smart"
},
"long_name":{
"type": "text",
"analyzer": "ik_smart"
},
"brand_id":{
"type": "integer"
},
"category_id":{
"type":"integer"
},
"shop_id":{
"type":"integer"
},
"price":{
"type":"scaled_float",
"scaling_factor":100
},
"sold_count":{
"type":"integer"
},
"review_count":{
"type":"integer"
},
"status":{
"type":"integer"
},
"create_time" : {
"type" : "date"
},
"last_time" : {
"type" : "date"
},
"skus":{
"type":"nested",
"properties": {
"name":{
"type":"text",
"analyzer":"ik_smart"
},
"price":{
"type":"scaled_float",
"scaling_factor":100
}
}
},
"attributes":{
"type":"nested",
"properties": {
"name":{
"type":"keyword"
},
"value":{
"type":"keyword"
}
}
}
}
}
}
索引规则解释:
"analyzer": "ik_smart" 代表这个字段需要使用 IK 中文分词器分词。
还有有一些字段的类型是 keyword，这是字符串类型的一种，这种类型是告诉 Elasticsearch 不需要对这个字段做分词，通常用于邮箱、标签、属性等字段。
scaled_float 代表一个小数位固定的浮点型字段，与 Mysql 的 decimal 类型类似，后面的 scaling_factor 用来指定小数位精度，100 就代表精确到小数点后两位。
skus 和 attribute 的字段类型是 nested，代表这个字段是一个复杂对象，由下一级的 properties 字段定义这个对象的字段。有人可能会问，我们的『商品 SKU』和『商品属性』明明是对象数组，为什么这里可以定义成对象？这是 Elasticsearch 的另外一个特性，每个字段都可以保存多个值，这也是 Elasticsearch 的类型没有数组的原因，因为不需要，每个字段都可以是数组。
input {
stdin { }
jdbc {
#注意mysql连接地址一定要用ip，不能使用localhost等
jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
jdbc_user => "dark"
jdbc_password => "mysql"
#数据库重连尝试
connection_retry_attempts => "3"
#数据库连接可用校验超时时间，默认为3600s
jdbc_validation_timeout => "3600"
#这个jar包的地址是容器内的地址
jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
#开启分页查询（默认是false）
jdbc_paging_enabled => "true"
#单次分页查询条数（默认100000，字段较多的话，可以适当调整这个数值）
jdbc_page_size => "50000"
#执行的sql语句
statement => "SELECT a.id,a.`name`,a.long_name,a.brand_id,a.three_category_id as category_id,a.shop_id,a.price,a.sold_count,a.review_count,a.`status`,a.create_time,a.last_time,b.`name` as category,b.path FROM lmrs_products as a LEFT JOIN lmrs_product_categorys as b ON a.three_category_id = b.id where a.id > :sql_last_value"
#需要记录查询结果某字段的值时，此字段为true，否则默认tracking_colum为timestamp的值
use_column_value => true
#是否将字段名转为小写，默认为true（如果具备序列化或者反序列化，建议设置为false）
lowercase_column_names => false
#需要记录的字段，同于增量同步，需要是数据库字段
tracking_column => id
#记录字段的数据类型
tracking_column_type => numeric
#上次数据存放位置
record_last_run => true
#上一个sql_last_value的存放路径，必须在文件中指定字段的初始值
last_run_metadata_path => "/etc/logstash/pipeline/products.txt"
#是否清除last_run_metadata_path的记录，需要增量同步这个字段的值必须为false
clean_run => false
#同步的频率（分 时 天 月 年）默认为每分钟同步一次
schedule => "* * * * *"
type => "_doc"
}
jdbc {
#注意mysql连接地址一定要用ip，不能使用localhost等
jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
jdbc_user => "dark"
jdbc_password => "mysql"
#数据库重连尝试
connection_retry_attempts => "3"
#数据库连接可用校验超时时间，默认为3600s
jdbc_validation_timeout => "3600"
#这个jar包的地址是容器内的地址
jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
#开启分页查询（默认是false）
jdbc_paging_enabled => "true"
#单次分页查询条数（默认100000，字段较多的话，可以适当调整这个数值）
jdbc_page_size => "50000"
#执行的sql语句
statement => "select c.*,d.`name`as category,d.path as category_path from (select b.id,a.`name`,b.`name` as `value`,a.sort as attribute_sort,b.sort as attribute_value_sort,a.category_id from lmrs_attributes as a LEFT JOIN lmrs_attribute_values as b on a.id = b.attribute_id) as c LEFT JOIN lmrs_product_categorys as d on c.category_id = d.id where c.id > :sql_last_value"
#需要记录查询结果某字段的值时，此字段为true，否则默认tracking_colum为timestamp的值
use_column_value => true
#是否将字段名转为小写，默认为true（如果具备序列化或者反序列化，建议设置为false）
lowercase_column_names => false
#需要记录的字段，同于增量同步，需要是数据库字段
tracking_column => id
#记录字段的数据类型
tracking_column_type => numeric
#上次数据存放位置
record_last_run => true
#上一个sql_last_value的存放路径，必须在文件中指定字段的初始值
last_run_metadata_path => "/etc/logstash/pipeline/attributes.txt"
#是否清除last_run_metadata_path的记录，需要增量同步这个字段的值必须为false
clean_run => false
#同步的频率（分 时 天 月 年）默认为每分钟同步一次
schedule => "* * * * *"
type => "attribute"
}
}
filter {
if [type] == "_doc"{
jdbc_streaming {
jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
jdbc_user => "dark"
jdbc_password => "mysql"
parameters => {"product_id"=>"id"}
statement => "select `name`,price from lmrs_product_skus where product_id = :product_id"
target => "skus"
}
jdbc_streaming {
jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
jdbc_user => "dark"
jdbc_password => "mysql"
parameters => {"product_id"=>"id"}
statement => "SELECT c.`name`,f.`name` as `value` FROM (SELECT a.name,a.id FROM lmrs_attributes as a LEFT JOIN lmrs_product_attribute_values as b on a.id = b.attribute_id WHERE b.product_id = :product_id) as c LEFT JOIN(SELECT d.attribute_id,d.name FROM lmrs_attribute_values as d LEFT JOIN lmrs_product_attribute_values as e ON d.id = e.attribute_value_id WHERE product_id = :product_id) as f ON c.id = f.attribute_id GROUP BY f.name"
target => "attributes"
}
}
}
output {
if [type] == "_doc" {
elasticsearch {
#注意es连接地址一定要用ip，不能使用localhost等
hosts => "172.17.0.7:9200"
index => "products"
document_type => "_doc"
document_id => "%{id}"
}
}
if [type] == "attribute" {
elasticsearch {
#注意es连接地址一定要用ip，不能使用localhost等
hosts => "172.17.0.7:9200"
index => "attribute"
document_type => "_doc"
document_id => "%{id}"
}
}
stdout {
codec => json_lines
}
}
解释：
filter 中 target主要对应 products索引中的两个嵌套字段（nested）
注意：
需要给两个txt文件相应的权限，详见单表操作
同步后的products索引数据
{
"took" : 552,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"skipped" : 0,
"failed" : 0
},
"hits" : {
"total" : {
"value" : 2,
"relation" : "eq"
},
"max_score" : 1.0,
"hits" : [
{
"_index" : "products",
"_type" : "_doc",
"_id" : "1",
"_score" : 1.0,
"_source" : {
"sold_count" : 111,
"skus" : [
{
"price" : 6299,
"name" : "皓月银 I5/16GB/512GB 触屏 集成显卡 官方标配"
},
{
"price" : 6599,
"name" : "皓月银 I7/16GB/512GB 触屏 集成显卡 官方标配"
},
{
"price" : 6299,
"name" : "深空灰 I5/16GB/512GB 触屏 集成显卡 官方标配"
},
{
"price" : 6599,
"name" : "深空灰 I7/16GB/512GB 触屏 集成显卡 官方标配"
},
{
"price" : 6299,
"name" : "樱粉金 I5/16GB/512GB 触屏 集成显卡 官方标配"
},
{
"price" : 6599,
"name" : "樱粉金 I7/16GB/512GB 触屏 集成显卡 官方标配"
}
],
"type" : "_doc",
"@version" : "1",
"price" : 6299.0,
"brand_id" : 1,
"attributes" : [
{
"value" : "皓月银",
"name" : "颜色"
},
{
"value" : "深空灰",
"name" : "颜色"
},
{
"value" : "樱粉金",
"name" : "颜色"
},
{
"value" : "I5/16GB/512GB 触屏",
"name" : "配置"
},
{
"value" : "I7/16GB/512GB 触屏",
"name" : "配置"
},
{
"value" : "集成显卡",
"name" : "显卡"
},
{
"value" : "官方标配",
"name" : "类型"
}
],
"shop_id" : 1,
"id" : 1,
"category_id" : 440,
"path" : "-425-438-",
"@timestamp" : "2021-05-31T09:19:01.109Z",
"last_time" : "2021-05-31T15:41:14.000Z",
"category" : "笔记本电脑",
"review_count" : 1111,
"long_name" : "HUAWEI Mate Book 13 16GB 512GB 触屏 集显",
"status" : 1,
"create_time" : "2021-05-25T15:12:09.000Z",
"name" : "HUAWEI Mate Book 13"
}
},
{
"_index" : "products",
"_type" : "_doc",
"_id" : "2",
"_score" : 1.0,
"_source" : {
"sold_count" : 222,
"skus" : [
{
"price" : 7999,
"name" : "翡冷翠 R5/32GB/1TB 触屏 触屏 集成显卡 官方标配"
},
{
"price" : 9999,
"name" : "翡冷翠 R7/32GB/1TB 触屏 触屏 集成显卡 官方标配"
},
{
"price" : 7999,
"name" : "冰霜银 R5/32GB/1TB 触屏 触屏 集成显卡 官方标配"
},
{
"price" : 9999,
"name" : "冰霜银 R7/32GB/1TB 触屏 触屏 集成显卡 官方标配"
},
{
"price" : 7999,
"name" : "星际蓝 R5/32GB/1TB 触屏 触屏 集成显卡 官方标配"
},
{
"price" : 9999,
"name" : "星际蓝 R7/32GB/1TB 触屏 触屏 集成显卡 官方标配"
}
],
"type" : "_doc",
"@version" : "1",
"price" : 7999.0,
"brand_id" : 2,
"attributes" : [
{
"value" : "翡冷翠",
"name" : "颜色"
},
{
"value" : "冰霜银",
"name" : "颜色"
},
{
"value" : "星际蓝",
"name" : "颜色"
},
{
"value" : "R5/32GB/1TB 触屏",
"name" : "配置"
},
{
"value" : "R7/32GB/1TB 触屏",
"name" : "配置"
},
{
"value" : "集成显卡",
"name" : "显卡"
},
{
"value" : "官方标配",
"name" : "类型"
}
],
"shop_id" : 2,
"id" : 2,
"category_id" : 440,
"path" : "-425-438-",
"@timestamp" : "2021-05-31T09:19:01.110Z",
"last_time" : "2021-05-31T21:10:04.000Z",
"category" : "笔记本电脑",
"review_count" : 222,
"long_name" : "HUAWEI Mate Book 14 32GB 1TB 触屏 集显",
"status" : 1,
"create_time" : "2021-05-28T20:02:02.000Z",
"name" : "HUAWEI Mate Book 14"
}
}
]
}
}
