高峰只对攀登它而不是仰望它的人来说才有真正意义。

已解决:logstash从MySQL导数据到es, 每次重启logstash,数据就重新导入

Logstash | 作者 wuminhe | 发布于2020年02月17日 | 阅读数:2263

环境:docker19.03 logstash6.5.4 es6.5.4 mysql8.0
logstash从MySQL导数据到es, 每次重启logstash,数据就重新导入,last_run_metadata_path也没有生成。请帮忙看看配置在哪里出现问题了。一直没有解决掉。
input {
jdbc {
jdbc_connection_string => "jdbc:mysql://localhost:3306/advertisement?useUnicode=true&characterEncoding=utf-8"
jdbc_user => "root"
jdbc_password => "root"

jdbc_driver_library => "/usr/share/logstash/config/jdbc/mysql-connector-java-8.0.18.jar"
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
jdbc_paging_enabled => "true"
jdbc_page_size => "5000"
jdbc_default_timezone => "Asia/Shanghai"
message_id AS messageId,
comment_id AS commentId,
order_num AS orderNum,
ori_author AS oriAuthor,
IFNULL(original_flag,0) AS originalFlag,
title,
summary,
site,
author,
DATE_FORMAT( public_time, '%Y-%m-%d %H:%i:%S') AS publicTime,
DATE_FORMAT( public_time, '%h:%i:%s') AS pTime,
clicks_count AS clicksCount,
like_count AS likeCount,
comments_count AS commentsCount,
ad_flag AS adFlag,
rumor_flag AS rumorFlag,
rewards_count AS rewardsCount,
url,
image_url AS imageUrl,
source_url AS sourceUrl,
source_username AS sourceUserName,
source_usernickname AS sourceUserNickName,
video_url AS videoUrl,
music_url AS musicUrl,
audio_url AS audioUrl,
memo,
download_status AS downloadStatus,
DATE_FORMAT( insert_time, '%Y-%m-%d %H:%i:%S') AS insertTime,
DATE_FORMAT( update_time, '%Y-%m-%d %H:%i:%S') AS updatetime,
account_id AS accountId,
wechat_id AS wechatId,
content,
content_length AS contentLength,
type
FROM w_data
WHERE
update_time >= :sql_last_value
AND clicks_count is not null
AND public_time >= '2019-10-01 00:00:00'
ORDER BY update_time ASC"

schedule => "* * * * *"

#record_last_run => true

use_column_value => true

tracking_column => "updatetime"

tracking_column_type => "timestamp"

last_run_metadata_path => "syncpoint_updatetime"

clean_run => "false"

lowercase_column_names => "false"

}
}
filter {
mutate {
add_field => { "index_date" => "%{publicTime}" }
}

mutate {
convert => [ "index_date", "string" ]
gsub => ["index_date", "-\d{2}\s\d{2}:\d{2}:\d{2}", ""]
gsub => ["index_date", "-", ""]
}
}

output {
elasticsearch {
hosts => [ "elasticsearch:9200" ]
index => "test-index"
document_type => "data"
document_id => "%{id}"
template_overwrite => true
}
}
已邀请:

super9du

赞同来自:

你把 `#record_last_run => true` 这个注释了啊,大哥

要回复问题请先登录注册