AWS의 Athena (AWS) 데이터베이스에서 병렬 읽기,

Question

AWS의 Athena 데이터베이스에서 largish 데이터 세트가 있습니다. 여기에 내가 노력하고 무엇을 나는 병렬 그것에서 읽고 싶은, 그리고 내가 RJDBC

을 사용하고

R.

내에서 분기에 foreach 패키지의 접근 방식에 익숙해있어 :

out <- foreach(i = 1:length(fipsvec), .combine = rbind, .errorhandling = "remove") %dopar% { 
    coni <- dbConnect(driver, "jdbc:awsathena://<<location>>/", 
      s3_staging_dir="my_directory", 
      user="...", 
      password="...") 
    print(paste0("starting ", i)) 
    sqlstring <- paste0("SELECT ", 
      "My_query_body" 
      fipsvec[i] 
    ) 
    row <- fetch(dbSendQuery(coni, sqlstring), -1, block = 999)   
    print(i) 
    dbDisconnect(coni) 
    rm(coni) 
    gc() 
    return(row) 
} 

가 (미안이 재현 할 수 없습니다 -. 내가 분명히 온라인 DB의 열쇠를 손 수) 나는이 프로그램을 실행할 때

가 첫 번째 c = number of cores 단계는 잘 실행하지만, 다음이 정지하고 아무것도하지 않는다 - indefi 니딜리 내가 말할 수있는 한. htop은 모든 코어에서 아무런 활동도 보이지 않습니다. 그리고 for 루프가 c 개 이상의 루프로 바뀌면 결과는 기대 한 것입니다. 병렬에서 직렬로 변경하면 (대신 %dopar%) 잘 작동합니다.

연결이 제대로 닫히지 않았거나 중복 된 것으로 정의되는 것과 관련이 있습니까? 병렬 루프 내에 연결을 배치 했으므로 각 코어는 자체 환경에서 자체 연결을 가져야합니다. 하지만 충분히 명확한 지 여부를 알기 위해 데이터베이스에 대해 충분히 알지 못합니다.

여기에 어떤 일이 벌어지고 있는지 이해하는 데 도움이되는 답변을 보내 주시면 감사하겠습니다.이 시점에서 나에게 부두입니다.

Answer 1

RJDBC 패키지 (및 종속성 - methods, DBI 및 rJava)를 클러스터에 전달하면됩니까?

하지 않으면 것은, 코드의 당신의 첫 번째 줄은 다음과 같이 보일 것이다 : 나는 생각한다 (그러나 모르는)

results <- foreach(i = 1:length(fipsvec), 
        .combine = rbind, 
        .errorhandling = "remove", 
        .packages=c('methods','DBI','rJava','RJDBC')) %dopar% { 

한 가지 일을 할 수있는 좀 더 털이가 RJDBC은을 사용한다는 것입니다 쿼리를 실행할 JVM rJava가 JVM 초기화를 처리하는 방법과 각 스레드가 동일한 JVM을 동시에 재사용하려고하는 경우 또는 처음부터 올바르게 초기화하기 위해 외부 환경에 대한 충분한 정보가있는 경우에 대해 잘 알고 있지 않습니다.

위의 방법으로 문제가 해결되지 않으면 driver의 할당 단계를 %dopar% 환경으로 옮길 수도 있습니다.

다른 트랙에서는 결과 집합에 몇 개의 행이 있습니까? 결과 집합이 백만 개가 넘는 행 범위에 있고 단일 쿼리로 반환 될 수 있다면 사실 RJDBC 패키지 내에서 최적화 할 수있는 기회를 얻었으며 github (https://github.com/s-u/RJDBC/pull/50)에 대한 공개 요청이있어 몇 달 동안 나 자신을 사용 해왔다. 끌어 오기 요청에 문서화 된 기본 벤치 마크가 있습니다. 실행중인 특정 쿼리에서 속도 향상이 상당하다는 것을 알았습니다.

당신이 지점을 설치할 수 있습니다 적용 할 것 같으면 :

library(devtools)  
devtools::install_github("msummersgill/RJDBC",ref = "harmonize", force = TRUE)