logstash postgres to elasticsearch

tl;dr

pgsync를 써볼걸 그랬나? https://github.com/toluaina/pg-sync

상황 

- postgres에서 elasticsearch로 데이터 동기화 하려고함

- geometry, json, geo_shape 타입 처리

- 언제나 그렇듯 개발서버에서는 문제없는데 운영에서는 제약이 있고, 내가 설치한 것도 아니라 디렉토리 권한 같은 것도 하나하나 확인하며 실행시켜야하죠

 

해결방법

디렉토리는 내기준 && 수정한 부분

 

- cd /usr/share/logstash 에다가 data2 폴더 만들어서했음.

(필수적인 부분 아님)

 

- vi /etc/logstash/jvm.options

## JVM configuration

-Xms12g
-Xmx12g

(heap size 늘리기위해서)

 

- vi /etc/systemd/system/logstash.service

/etc/systemd/system/logstash.service

ExecStart=/usr/share/logstash/bin/logstash "-f" "/etc/logstash/conf.d/내가만든파일이름whatever.conf" "--path.data" "/usr/share/logstash/data2"

이 부분(ExecStart)를 수정해주어야 background job으로 logstash를 실행할 수 있다. 

수정후에는 

sudo systemctl daemon-reload

해줘야함 

 

 

- 위에 언급된 whatever~ blahblah.conf 파일 같은 경우에는 

input {
	jdbc {
    	jdbc_connection_string => "jdbc:postgresql://IP:HOST/DBNAME"
        jdbc_user => "계정"
        jdbc_password => "비밀번호"
        jdbc_driver_library => "다운받는버전에맞는driver.jar"
        jdbc_validate_connection => true
        jdbc_driver_class => "org.postgresql.Driver"
		columns_charset => {"region_name_kr" => "UTF-8"}
        schedule => "*/15 * * * *"
        statement => "SELECT 어쩌고저쩌고들. 떨거지 컬럼들... geometry컬럼의 경우에는::text로 해서 받기 from table order by 기준컬럼 asc"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "200000"
    }
    stdin {
    	codec => plain { charset => "UTF-8"}
    }
}
filter {
	json { source => "continent" }
    json { source => "country" }
}
output {
	elasticsearch {
    	hosts => ["https://도메인:443"]
        index => "인덱스이름"
        doc_as_upsert => true
        action => "update"
        document_id => "%{region_id}"
    }
    stdout { codec => rubydebug }
}

이 파일은 json 인 경우에 json으로 받지 못하는 문제점이 아직 있음.

jdbc_paging_enabled, jdbc_page_size 옵션이 있어야 실행됨.

 

 

 

아래 파일이 json 해결 된 버전.

input {
    jdbc {
        jdbc_connection_string => "jdbc:postgresql://localhost:5432/atlasdb?useTimezone=true&useLegacyDatetimeCode=false&serverTimezone=UTC&useSSL=false&useUnicode=true&characterEncoding=utf8"
        jdbc_user => "atlas"
        jdbc_password => "atlas"
        jdbc_validate_connection => true
        jdbc_driver_library => "/lib/postgres-42-test.jar"
        jdbc_driver_class => "org.postgresql.Driver"
        schedule => "* * * * *"
        statement => "SELECT region_id, continent::text, country::text from expedia_region_union_copy_jpa_test order by region_id asc limit 100"
   }

   stdin {
        codec => plain { charset => "UTF-8"}
   }

}
filter {
 ruby {
        code => "
            require 'json'
            continent_json = JSON.parse(event.get('continent').to_s)
            event.set('continent', continent_json)
            country_json = JSON.parse(event.get('country').to_s)
            event.set('country', country_json)
        "
    }
}

output {
 elasticsearch {
        hosts => [ "localhost:9200" ]
        index => "2020-05-18-mon-001"
        doc_as_upsert => true
        action => "update"
        document_id => "%{region_id}"
        }
 stdout { codec => rubydebug }
}

---

geo_shape 매핑

 

결론

- 로그스태시는 설정이 다임. memory 체크 할것

습득 지식

- 설정은 어렵지만 할게 설정뿐이라면 즐겁다.

 

소요시간

- while 코로나 

 

---

참고 링크

- event filter ruby : https://www.elastic.co/guide/en/logstash/current/event-api.html

- jdbc_paging : github.com/logstash-plugins/logstash-input-jdbc/issues/9#issuecomment-119017577

- ExecStart :  https://discuss.elastic.co/t/logstash-service-doesnt-output-logs-command-line-does/90294/2