Równoległa pętla for w systemie Windows

Question

Wykonuję wiele zapytań SQL do bazy danych o dużej sile przemysłowej, ale otrzymanie wyników zajmuje dużo czasu. Było znacznie szybciej, gdy mój komputer z R był prawie obok bazy danych, co prowadzi mnie do przekonania, że ​​to opóźnienie między moim komputerem a bazą danych jest wąskim gardłem, a uruchamianie równoległych zapytań może przyspieszyć działanie. Jesteśmy na różnych kontynentach.

Oto wersja robocza, która nie jest równoległa:

doQueries <- function(filenameX, inp1, inp2) { 
    print(paste("Starting:", inp1, inp2, ",saving to", filenameX, sep=" ")) 
    # Here should the query be (using RODBC) 
    # save(queryresults, file="filenameX") 
} 

input.rows <- cbind(c("file1.rda","file2.rda","file3.rda"),c("A","B","C"),c(12,13,14)) 

for (i in 1:nrow(input.rows)) { 
    doQueries(filenameX=input.rows[i,1], inp1=input.rows[i,2], inp2=input.rows[i,3]) 
} 

próbowałem z następującego kodu, ale foreach Biblioteka wydają się nie być dostępna, a jak rozumiem od czytania na CRAN, parallel zastępuje wcześniejsze pakiety dla parallellingu ("pakiet" foreach "nie jest dostępny (dla R wersji 2.15.0)").

library(parallel) 
library(foreach) 
foreach (i=1:nrow(input.rows)) %dopar% { 
    doQueries(filenameX=input.rows[i,1], inp1=input.rows[i,2], inp2=input.rows[i,3]) 
} 

Jak powinienem to zrobić?

Podziękowania dla wszystkich współpracowników za Stackoverflow!

/Chris

Aktualizacja: Dzięki nograpes, udało mi się załadować bibliotek. Poniższy kod wydaje się działać:

library(RODBC) 
library(doParallel) 
library(foreach) 

# odbcCloseAll() 
# my_conn <- odbcConnect("database", uid="xx", pwd="yy", case="nochange") 

doQueries <- function(filenameX, inp1, inp2) { 
    print(paste("Starting:", inp1, inp2, ",saving to", filenameX, sep=" ")) 
    # sql.test <- RODBC::sqlQuery(my_conn, "SELECT * FROM zzz LIMIT 100", rows_at_time=1024) 
    # save(sql.test, file="filenameX") 
} 

input.rows <- cbind(c("file1.rda","file2.rda","file3.rda"),c("A","B","C"),c(12,13,14)) 

cl <- makeCluster(3) 
registerDoParallel(cl) 

foreach (i=1:nrow(input.rows)) %dopar% { 
    doQueries(filenameX=input.rows[i,1], inp1=input.rows[i,2], inp2=input.rows[i,3]) 
} 

stopCluster(cl) 

Ale kiedy to rzeczywiste zapytania SQL, pojawia się ten błąd, komunikat: błąd w {: Zadanie 1 nie powiodło się - „pierwszy argument nie jest otwarty kanał RODBC”

Czy to możliwe, że koncepcyjnie nie zadziała? Że RODBC nie może obsłużyć więcej niż jednego zapytania naraz?

Naprawdę doceniam wsparcie.

/Chris

Aktualizacja 2: Dzięki a nograpes wiele dla bardzo dobrych i imponujących odpowiedzi. Trudno jest ocenić, czy same transfery danych są szybsze (myślę o około 20% szybszej całkowitej przepustowości), ale odkryłem, że ponieważ zapytania (około 100) mają różne czasy odpowiedzi i wymagają postprocessingu (który włączam do funkcji przed zapisaniem), uzyskuję lepsze wykorzystanie połączenia i lokalnego procesora. To znaczy. za pomocą tylko jednej kwerendy w tym czasie, procesory będą prawie nieużywane podczas przesyłania danych, a następnie łącze będzie ciche podczas pracy procesorów. W przypadku zapytań równoległych widzę dane przychodzące i procesory pracujące w tym samym czasie. W sumie stało się to znacznie szybciej. Wielkie dzięki!

/Chris

Answer 1

Jak wspomniałem w moim komentarzu, technika ta prawdopodobnie nie będzie szybciej. Aby odpowiedzieć na twoje pytanie, pakiet foreach jest dostępny dla twojej wersji R. Być może wybrane repozytorium nie zostało jeszcze zaktualizowane. Wypróbuj to:

install.packages('foreach', repos='http://cran.us.r-project.org') 

, który powinien zainstalować pakiet. Jeśli to nie zadziała, pobierz plik binarny dla swojego systemu operacyjnego here i zainstaluj go w menu.

Jeśli wąskim gardłem jest połączenie sieciowe, można jedynie przyspieszyć proces, zmniejszając ilość materiałów wprowadzanych do sieci. Jednym z pomysłów byłoby zdalne połączenie z serwerem bazy danych, wysłanie zapytania do pliku (na serwerze), skompresowanie go, a następnie pobranie go na komputer, a następnie rozpakowanie R i załadowanie go.To brzmi jak dużo, ale prawdopodobnie możesz zrobić cały proces w R.

Kontynuując aktualizację, okazuje się, że nie dodano argumentu .packages w swoim oświadczeniu foreach. Właśnie dlatego musieli Państwo wstępnie ustawić funkcję sqlQuery na RODBC::. Konieczne jest określenie, jakie pakiety potrzebuje ta pętla, ponieważ myślę, że zasadniczo rozpoczyna ona nową sesję R dla każdego węzła i każda sesja musi zostać zainicjowana pakietami. Podobnie, nie możesz uzyskać dostępu do my_conn, ponieważ był poza pętlą, musisz utworzyć wewnątrz pętli, aby każdy węzeł miał swoją własną kopię.

library(RODBC) 
library(foreach) 
library(doParallel) 
setwd('C:/Users/x/Desktop') 

doQueries <- function(filenameX) { 
    sql.text<-sqlQuery(my_conn, 'SELECT * FROM table;') 
    save(sql.text, file=filenameX) 
} 

cl <- makeCluster(2) 
registerDoParallel(cl) 

foreach (i=1:2, .packages='RODBC') %dopar% { 
    my_conn <- odbcConnect("db", uid="user", pwd="pass") 
    doQueries(filenameX=paste('file_',i,sep='')) 
} 

Ale, jak wspomniałem, to prawdopodobnie nie będzie szybsze.