0Pricing
R Academy · Ders

Paralel Kodda Hata Ayıklama ve Yük Dengeleme

Paralel çalışanlardaki hataları yönetin ve eşit olmayan iş yüklerini etkili biçimde dengeleyin.

Paralel Kodda Hata Ayıklama ve Yük Dengeleme, CoddyKit'te ücretsiz bir R Academy dersidir. Bu, 4 dersinin 4. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, R Academy öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. R Academy kursu toplamda 4 dersten oluşur.

Paralel Hata Ayıklama Neden Zordur?

Paralel kodda hata ayıklamak zordur; çünkü çalışan işlemler ayrı süreçlerde çalışır: print() ifadeleri ana işlemde görünmez, browser() gibi etkileşimli hata ayıklayıcılar çalışan işlemlerin içinde çalışmaz ve hatalar serileştirilip ana işlemde yeniden oluşturulduğundan özgün çağrı yığını kaybolur.

library(parallel)

# Problem: cat() inside worker is invisible in the master
cl <- makeCluster(2)
clusterExport(cl, c())

result <- parLapply(cl, 1:4, function(i) {
  # This cat() goes to the worker's stdout, NOT the master
  cat('Worker processing', i, '\n')  # you will NOT see this
  i^2
})

cat('Master received:', unlist(result), '\n')
stopCluster(cl)

parLapply'de Hataların Yayılması

Bir çalışan işlem hata oluşturduğunda parLapply() hatayı sarar ve ana işlemde yeniden oluşturur. Hata iletisi korunur; ancak uzak çağrı yığını korunmaz. İşlevinizi paralelleştirmeden önce her zaman sıralı olarak test edin.

library(parallel)
cl <- makeCluster(2)

# Step 1: test sequentially first!
my_fn <- function(x) {
  if (x == 3) stop('bad value: 3')
  sqrt(x)
}

# Sequential test reveals the bug before parallelising
tryCatch(
  lapply(1:5, my_fn),
  error = function(e) cat('Sequential error caught:', e$message, '\n')
)

# Fix: guard the bad case
my_fn_safe <- function(x) {
  if (x <= 0) return(NA_real_)
  sqrt(x)
}
clusterExport(cl, 'my_fn_safe')
cat(unlist(parLapply(cl, 1:5, my_fn_safe)), '\n')

stopCluster(cl)

Çalışan İşlevlerin İçinde tryCatch

Çalışan işlevinin gövdesini tryCatch() içine almak, tüm paralel görevi çöktürmeden her öğeye ait hataları yakalamanızı ve günlüğe kaydetmenizi sağlar. Başarısızlık durumunda NA gibi bir gösterge değer döndürün; böylece işlem sonrası adım sorunlu girdileri belirleyebilir.

library(parallel)
cl <- makeCluster(2)

safe_compute <- function(x) {
  tryCatch({
    if (x == 3) stop('simulated error on element 3')
    list(value = x^2, error = NULL)
  }, error = function(e) {
    list(value = NA_real_, error = conditionMessage(e))
  })
}
clusterExport(cl, 'safe_compute')

results <- parLapply(cl, 1:5, safe_compute)

for (i in seq_along(results)) {
  r <- results[[i]]
  if (is.null(r$error)) {
    cat('Element', i, ': value =', r$value, '\n')
  } else {
    cat('Element', i, ': ERROR -', r$error, '\n')
  }
}

stopCluster(cl)

tryCatch ile future::value()

future paketinde value(f), uzak hatayı yeniden oluşturur. Bunu tryCatch() içine almak, diğer future'lardan sonuç toplamaya devam ederken tek tek future başarısızlıklarını ele almanızı sağlar.

library(future)
plan(multisession, workers = 2)

# Create futures — some will fail
inputs <- list(4, -1, 9, 'x', 16)

futures <- lapply(inputs, function(val) {
  future({
    if (!is.numeric(val)) stop('non-numeric input')
    if (val < 0) stop('negative input')
    sqrt(val)
  })
})

# Collect with individual error handling
results <- lapply(futures, function(f) {
  tryCatch(
    value(f),
    error = function(e) paste('ERROR:', e$message)
  )
})

for (i in seq_along(results)) {
  cat('Input:', inputs[[i]], '-> Result:', as.character(results[[i]]), '\n')
}

plan(sequential)

foreach ile %dopar% ve .errorhandling

foreach paketinin %dopar% işleci, yinelemeleri kayıtlı bir arka uç arasında dağıtır. .errorhandling bağımsız değişkeni hatalarda ne olacağını denetler: 'stop' (varsayılan), 'remove' (atla) veya 'pass' (koşul nesnesini dahil et).

library(foreach)
library(doParallel)

cl <- makeCluster(2)
registerDoParallel(cl)

# .errorhandling = 'pass': failed elements return the condition
results <- foreach(
  x = 1:6,
  .errorhandling = 'pass'
) %dopar% {
  if (x == 4) stop('bad element')
  x^2
}

for (i in seq_along(results)) {
  if (inherits(results[[i]], 'error')) {
    cat('Element', i, ': ERROR -', results[[i]]$message, '\n')
  } else {
    cat('Element', i, ': value =', results[[i]], '\n')
  }
}

stopCluster(cl)

Yük Dengeleme: Statik ve Dinamik

Statik zamanlama, eşit büyüklükteki parçaları çalışan işlemlere önceden atar. Dinamik zamanlama ise her çalışan işleme aynı anda bir görev verir; böylece hızlı çalışan işlemler daha fazla görev alır. Görev süreleri büyük ölçüde değişiyorsa dinamik zamanlama daha iyidir.

library(parallel)
cl <- makeCluster(2)

# Simulate unequal task durations (element i takes i*0.05 seconds)
unequal_task <- function(i) {
  Sys.sleep(i * 0.05)
  i
}
clusterExport(cl, 'unequal_task')

# Static: parLapply distributes in fixed chunks
static_time <- system.time(
  parLapply(cl, 1:6, unequal_task)
)[['elapsed']]

# Dynamic: clusterApplyLB assigns one task per available worker
dynamic_time <- system.time(
  clusterApplyLB(cl, 1:6, unequal_task)
)[['elapsed']]

cat('Static LB: ', round(static_time, 2), 's\n')
cat('Dynamic LB:', round(dynamic_time, 2), 's\n')

stopCluster(cl)

Parçalara Bölme: Ek Yükü Azaltma

Süreçler arası iletişim, görev başına sabit bir ek yüke sahiptir. Çok sayıda küçük öğeyi işlerken bunları daha büyük parçalarda gruplayın; böylece her çalışan çağrısı ileti başına daha fazla iş yapar ve ek yükün hesaplamaya oranı azalır.

library(parallel)
cl <- makeCluster(2)

# Naive: 1000 tiny tasks — high overhead
tiny_task <- function(x) x^2
clusterExport(cl, 'tiny_task')

t1 <- system.time(parLapply(cl, 1:1000, tiny_task))[['elapsed']]

# Chunked: 10 tasks of 100 items each — low overhead
chunk_task <- function(chunk) sapply(chunk, function(x) x^2)
chunks <- split(1:1000, cut(1:1000, 10, labels = FALSE))
clusterExport(cl, 'chunk_task')

t2 <- system.time(parLapply(cl, chunks, chunk_task))[['elapsed']]

cat('Unchunked:', round(t1, 4), 's\n')
cat('Chunked:  ', round(t2, 4), 's\n')

stopCluster(cl)

Büyük Nesneleri Göndermekten Kaçının

Büyük nesneleri (veri çerçeveleri, matrisler, modeller) çalışan işlemlere serileştirmek pahalıdır. Bunun yerine yalnızca indisleri iletin ve verileri paylaşılan bir kaynaktan (dosya, veritabanı veya çalışan işlem içinde önceden yüklenmiş bir kaynak) okuyun. Çalışan işlemlere gönderilen yükü küçük tutun.

library(parallel)
cl <- makeCluster(2)

# BAD: sends the entire large data frame to every worker
big_df <- data.frame(x = rnorm(1e5), y = rnorm(1e5))
clusterExport(cl, 'big_df')  # 800 KB shipped to each worker

# BETTER: workers generate/read their own data slice
clusterEvalQ(cl, {
  set.seed(Sys.getpid())  # unique per worker
  local_data <- data.frame(x = rnorm(500), y = rnorm(500))
})

# Each worker uses its own local_data without receiving it from master
result <- parLapply(cl, 1:2, function(i) {
  coef(lm(y ~ x, data = local_data))
})
print(result)

stopCluster(cl)

Çalışan İşlemlerden Günlük Kaydı

Çalışan işlemler ana konsola yazamadığından, çalışan çıktısını makeCluster(outfile = '/path/to/log') kullanarak çalışan başına ayrı günlük dosyalarına yönlendirin. Unix'te çıktıyı /dev/null'a veya her çalışan için ayrılmış bir günlük dosyasına yönlendirebilirsiniz.

library(parallel)

# Route all worker stdout/stderr to a log file
log_file <- tempfile(fileext = '.log')
cl <- makeCluster(2, outfile = log_file)

clusterEvalQ(cl, {
  cat('[Worker', Sys.getpid(), '] started\n')
})

parLapply(cl, 1:4, function(i) {
  cat('[Worker] processing item', i, '\n')  # goes to log file
  i * 10
})

stopCluster(cl)

# Read the log
log_content <- readLines(log_file)
cat(head(log_content, 8), sep = '\n')

Paralel Kodu Profilleme

Toplam geçen süreyi ölçmek için system.time() kullanın ve çalışan işlemin ek yükünü hesaplama süresinden elle ayırın. Daha ayrıntılı inceleme için hedef işlevi önce profvis::profvis() ile sıralı olarak çalıştırın, ardından darboğazı paralelleştirin.

library(parallel)

# Profile the task sequentially first
target_fn <- function(n) {
  x <- rnorm(n)
  list(
    mean = mean(x),
    sd   = sd(x),
    q95  = quantile(x, 0.95)
  )
}

# Measure sequential baseline
t_seq <- system.time(lapply(rep(1000, 20), target_fn))[['elapsed']]

# Measure parallel speedup
cl <- makeCluster(2)
clusterExport(cl, 'target_fn')
t_par <- system.time(parLapply(cl, rep(1000, 20), target_fn))[['elapsed']]
stopCluster(cl)

cat('Sequential:', round(t_seq, 4), 's\n')
cat('Parallel:  ', round(t_par, 4), 's\n')
cat('Efficiency:', round(t_seq / (t_par * 2) * 100, 1), '%\n')

Eksiksiz Sağlam Düzen

Tüm iyi uygulamaları birleştirin: işi parçalara bölün, yük dengeli atama kullanın, hataları öğe başına ele alın, günlükleri dosyaya yazın ve on.exit() ile kümenin temizlenmesini güvence altına alın.

library(parallel)

robust_parallel <- function(items, fn, workers = 2) {
  cl <- makeCluster(workers, outfile = tempfile())
  on.exit(stopCluster(cl), add = TRUE)

  clusterExport(cl, 'fn', envir = environment())

  safe_fn <- function(x) {
    tryCatch(
      fn(x),
      error = function(e) list(error = e$message, value = NA)
    )
  }
  clusterExport(cl, 'safe_fn', envir = environment())

  # Load-balanced assignment for unequal task times
  clusterApplyLB(cl, items, safe_fn)
}

results <- robust_parallel(1:8, function(x) {
  if (x == 5) stop('deliberate error')
  x^3
})

cat('Results:\n')
for (i in seq_along(results)) {
  cat(' [', i, ']', ifelse(is.na(results[[i]]$value),
    paste('ERROR:', results[[i]]$error),
    results[[i]]
  ), '\n')
}

Hızlı Denetim

Çalışma süreleri son derece değişken olan 100 paralel göreviniz var (bazıları 10 ms, bazıları 500 ms sürüyor). Toplam geçen süreyi hangi zamanlama stratejisi en aza indirir?

Özet: Paralel Kodda Hata Ayıklama

Temel çıkarımlar:

  • Paralelleştirmeden önce işlevleri sıralı olarak test edin — önce lapply() kullanın
  • Görevi çökertmeden öğe başına hataları yakalamak için çalışan işlevlerin gövdelerini tryCatch() içine alın
  • tryCatch() ile birlikte kullanılan future::value(), tek tek future başarısızlıklarını düzgün biçimde ele alır
  • .errorhandling = 'pass' ile kullanılan foreach %dopar%, sonuç listesinde hata nesnelerini döndürür
  • Süreleri eşit olmayan görevlerde dinamik yük dengeleme için clusterApplyLB() kullanın
  • İletişim ek yükünü azaltmak için küçük görevleri parçalara ayırın
  • Çalışan işlemlere gönderilen yükü en aza indirin — büyük veri nesneleri yerine indisleri iletin
  • Çalışan çıktısını makeCluster(outfile=) aracılığıyla dosyalara kaydedin
# Canonical debugging workflow
# 1. Test sequentially
# lapply(inputs, my_fn)

# 2. Wrap in tryCatch
# safe_fn <- function(x) tryCatch(my_fn(x), error = function(e) NA)

# 3. Parallelise the safe version
# cl <- makeCluster(2); on.exit(stopCluster(cl))
# parLapply(cl, inputs, safe_fn)

cat('Debugging workflow: sequential -> tryCatch -> parallel\n')

Sıkça Sorulan Sorular

“Paralel Kodda Hata Ayıklama ve Yük Dengeleme” dersi ücretsiz mi?

Evet — “Paralel Kodda Hata Ayıklama ve Yük Dengeleme” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve R Academy kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. R Academy kursu toplamda 4 dersten oluşur.

“Paralel Kodda Hata Ayıklama ve Yük Dengeleme” dersinde ne öğreneceğim?

Paralel çalışanlardaki hataları yönetin ve eşit olmayan iş yüklerini etkili biçimde dengeleyin. R Academy ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.

R Academy öğrenmeye başlamak için deneyim gerekli mi?

Önceden deneyim gerekmez. CoddyKit'te R Academy, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 4. dersidir.

“Paralel Kodda Hata Ayıklama ve Yük Dengeleme” dersi ne kadar sürer?

Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.

Bu R Academy dersinde kod yazıp çalıştırabilir miyim?

Evet. Her R Academy dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.

Bu kursun tüm dersleri

  1. parallel Paketi ve detectCores()
  2. future Çerçevesi
  3. furrr: Paralel purrr İşlemleri
  4. Paralel Kodda Hata Ayıklama ve Yük Dengeleme
← R Academy Sayfasına Dön