انتقال داده ها از 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 => "/usr/share/logstash/mysql-connector-java-8.0.16.jar" 

    jdbc_driver_class => "com.mysql.jdbc.Driver" 

    jdbc_connection_string => "jdbc:mysql://SERVER_ADDRESS:PORT/DATABASE?autoReconnect=true&useSSL=false"

    jdbc_user => "USERNAME"

    jdbc_password => "PASSWORD"

    jdbc_paging_enabled => true

    tracking_column => "unix_ts_in_secs"

    use_column_value => true

    tracking_column_type => "numeric"

    schedule => "0 */60 * * * *"
    last_run_metadata_path => "/usr/share/logstash/sql_last_value.yml"

    statement => "SELECT *, 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"

  }

}

filter {

  mutate {

    copy => { "id" => "[@metadata][_id]"}

    remove_field => ["id", "@version", "unix_ts_in_secs"]

  }

}

output {

  elasticsearch {

      hosts => ["elasticsearch:9200"]

      #user => "elastic"

      #password => "PASSWORD"

      index => "INDEX_NAME"

      document_id => "%{[@metadata][_id]}"

  }

}

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

docker restart logstash

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

docker logs -f --tail 20 logstash

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