logstach : jdbc_page_size가 모든 데이터를 탄성 검색에 덤프하지 않습니다.

Question

tom_test2 postgresql 테이블을 탄성 검색으로 내보내고 싶습니다.

GET tom/tom_test/_search 

    "hits": { 
    "total": 176805, 
    "max_score": 1, 
} 

내 인덱스를 삭제하고 있습니다 : 탄성 검색

=> select count(*) from tom_test2; 
count 
-------- 
176805 
(1 row) 

다음 logstach의 conf 파일 가져 오기가 제대로 내 데이터 : 탄성 검색에서

input { 
    jdbc { 
     # Postgres jdbc connection string to our database, mydb 
     jdbc_connection_string => "xxx" 
     # The user we wish to execute our statement as 
     jdbc_user => "xxx" 
     jdbc_password => "xxx" 
     # The path to our downloaded jdbc driver 
     jdbc_driver_library => "xxx" 
     # The name of the driver class for Postgresql 
     jdbc_driver_class => "org.postgresql.Driver" 
     # our query 
     statement => "select * from tom_test2" 
    } 
} 


output { 
    elasticsearch { 
     hosts => ["xxx"] 
     index => "tom" 
     document_type => "tom_test" 
    } 
} 

표는 176,805 행이 탄성 검색에서 :

delete tom 

그리고 지금 내 데이터가 커지게 경우 jdbc_page_size를 사용하여 동일한 작업을 수행하고자하는 내 logstach의 conf 파일은 지금 :

GET tom/tom_test/_search 

    "hits": { 
    "total": 106174, 
    "max_score": 1, 
} 

176805로 :

input { 
    jdbc { 
     # Postgres jdbc connection string to our database, mydb 
     jdbc_connection_string => "xxx" 
     # The user we wish to execute our statement as 
     jdbc_user => "xxx" 
     jdbc_password => "xxx" 
     # The path to our downloaded jdbc driver 
     jdbc_driver_library => "xxx" 
     # The name of the driver class for Postgresql 
     jdbc_driver_class => "org.postgresql.Driver" 
     # our query 
     statement => "select * from tom_test2" 

     jdbc_page_size => 1000 
     jdbc_paging_enabled => true 
    } 
} 


output { 
    elasticsearch { 
     hosts => ["xxx"] 
     index => "tom" 
     document_type => "tom_test" 
    } 
} 

내 카운트가 지금 잘못 -106174 = 70631 개의 행이 누락되었습니다

Answer 1

이 문제에 직면 한 이유는 질의가 데이터 수신 순서를 제어하지 않기 때문입니다. 일반적으로 postgresql은 정렬되지 않은 후속 페이징에서이를 보장해서는 안됩니다 너를 불렀다. 동일한 데이터를 가져 오지 않습니다 : 일부 데이터가 전혀 가져 오지 않을 때 상황이 발생하고 일부 데이터는 여러 번 가져옵니다. (이러한 호출 중에 데이터가 수정되지 않더라도 백그라운드 진공 작업자가 실제 파일의 데이터를 분석하여 설명 된 상황을 재생산합니다.

진술 문에 SELECT * FROM tom_test2 ORDER BY id을 추가하고 데이터를 페이지에 추가하십시오. 하지만주의하십시오.이 경우 elasticsearch에 업로드하면 순간의 테이블 복제본을 정확히 확인할 수 없습니다. 그 원인은, 그 결과 페이징 요청의 로그 샷 처리 중에 곧 나오는 페이지의 데이터 업데이트가 도입된다는 것입니다. 즉, 페이지 1에서 10000으로 업로드하고 10001과 20000 페이지의 데이터에서 업데이트가 발생하고, 그렇지 않으면 나중에 발생합니다 ... 데이터의 일관성에 문제가 있습니다.

또는 모든 데이터를 가져 와서 logstash ...에서 메모리를 아낌없이 사용하려면 jdbc_fetch_size 매개 변수를 제어해야합니다. 즉, 동일한 SELECT * FROM tom_test2을 수행하고 있습니다. 이 접근 방식을 사용하면 하나의 쿼리 결과 집합을 만들지 만 조각으로 "펌핑"하고 "펌핑"중에 데이터 수정을 수행해도 쿼리가 시작될 때 상태가 페치됩니다.

Answer 2

의 경고로 jdbc_page_size의 쿼리간에 정렬이 보장되지 않기 때문에

큰 결과 집합에 documentation also says that으로 jdbc_page_size을 사용하는 대신 jdbc_fetch_size을 사용하는 것이 좋습니다.

P.S : 때때로) http://discuss.elastic.co에 당신의 질문을 더 logstash 탄성 관리자들에 의해