انتقال داده ها از MySQL به Elasticsearch

امروز یه چالش باحال داشتم و گفتم باهاتون به اشتراک بزارم شاید کارتون راحت تر و سریع تر انجام بشه.

برای انتقال داده ها از logstash استفاده میکنیم البته اگر از django استفاده میکنید نگاهی به این package ها هم بندازید. برای node کتابخونه های مشابهی نمیشناسم و سرچ هم نکردم البته اگر می شناسید معرفی کنید .o.O

خب کاملا واضحه که باید سرویس elasticsearch در حال اجرا باشه . من برای راحتی الستیک رو با داکر راه اندازی کردم .

docker run -d -p  9200:9200 --name elasticsearch elasticsearch:7.6.2

حالا باید سرویس logstash رو اجرا کنیم

docker run -d -v ~/pipeline:/usr/share/logstash/pipeline/  --link elasticsearch --name logstash logstash:7.6.2

برای ارتباط بین logstash و mysql از mysql-connector استفاده می کنیم که پیش فرض روی logstash وجود نداره به همین دلیل :

docker exec -itu 0 logsatsh bash

$ yum install -y wget
$ cd /usr/share/logstash
$ wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.20/mysql-connector-java-8.0.20.jar
$ cp mysql-connector-java-8.0.20.jar  /usr/share/logstash/logstash-core/lib/jars/
$ exit

حالا به مسیر ~/pipeline برید و فایل migrate.conf را بسازید.

input {

  jdbc {

    jdbc_driver_library => &quot/usr/share/logstash/mysql-connector-java-8.0.16.jar&quot 

    jdbc_driver_class => &quotcom.mysql.jdbc.Driver&quot 

    jdbc_connection_string => &quotjdbc:mysql://SERVER_ADDRESS:PORT/DATABASE?autoReconnect=true&useSSL=false&quot

    jdbc_user => &quotUSERNAME&quot

    jdbc_password => &quotPASSWORD&quot

    jdbc_paging_enabled => true

    tracking_column => &quotunix_ts_in_secs&quot

    use_column_value => true

    tracking_column_type => &quotnumeric&quot

    schedule => &quot0 */60 * * * *&quot
    last_run_metadata_path => &quot/usr/share/logstash/sql_last_value.yml&quot

    statement => &quotSELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC&quot

  }

}

filter {

  mutate {

    copy => { &quotid&quot => &quot[@metadata][_id]&quot}

    remove_field => [&quotid&quot, &quot@version&quot, &quotunix_ts_in_secs&quot]

  }

}

output {

  elasticsearch {

      hosts => [&quotelasticsearch:9200&quot]

      #user => &quotelastic&quot

      #password => &quotPASSWORD&quot

      index => &quotINDEX_NAME&quot

      document_id => &quot%{[@metadata][_id]}&quot

  }

}

در آخر کافیه سرویس رو restart کنید

docker restart logstash

می تونید لاگ ها رو هم نگاه کنید تا از نتیجه کار مطمن بشید

docker logs -f --tail 20 logstash

برای توضیحات بیشتر حتما این مقالات رو بخونید و تفاوت هایی کانفیگ بالا و مقالات زیر مطالعه کنید.