小型PCクラスタを用いた並列計算環境の構築 --- （必要に応じて、文脈や詳細情報があればさらに自然な表現へ調整できます。）

小さなパソコンでクラスタを構築して並列計算のプロセスを学ぶことができて楽しかったです。
Ubuntu のインストール、パスワードレス SSH、ノード間でのパッケージ自動化、R シミュレーションの分散実行、および CV5 と CV10 での性能比較に関するメモを書き留めました。面白いプロジェクトです！

動機

今年の目標の一つは、並列計算をさらに深掘りし、複数デバイスへシミュレーションワークロードを分散させることです。
ノートパソコンで何日もかけて実行されるシミュレーションは非効率的です。
クラウドサービスを試したものの、分散コアがどのように管理・最適化されているのか明確に理解できませんでした。
このプロジェクトでは、安価な「tiny PC」（例：中古 Lenovo M715q）で自前のクラスターを構築し、実際に手を動かす経験を得ます。

目的

1. ハードウェア – 適切な PC を選定
2. 各ノードへ Ubuntu をインストール
3. 静的 IP とパスワードレス SSH の設定
4. R のインストールとパッケージ配布を自動化
5. シミュレーション用のテンプレート R スクリプト作成
6. 各ノードへシミュレーション作業（開始/停止範囲）を分散
7. tmux セッション経由で並列実行
8. 結果収集、ランタイムと性能指標（バイアス・分散・カバレッジ）の比較

1. どの PC を購入するか？

機能的でコストパフォーマンスが高いユニットを選びます。中古 Lenovo M715q Tiny PC が最適です。

2. Ubuntu のインストール

結果：Ubuntu が迅速かつ円滑にインストールされます。

3. IP の固定と設定

ルーターで各ノード用に静的 IP を設定：

192.168.1.101
192.168.1.102
192.168.1.103

IP 設定変更後はノードを再起動します。

4. パスワードレス SSH

# マスターノードで実行
ssh-keygen -t rsa -b 4096 -f ~/.ssh/id_rsa

# 各ノードへキーをコピー（IP とユーザー名は置き換え）
for ip in 192.168.1.{101..103}; do
    ssh-copy-id -i ~/.ssh/id_rsa.pub user@$ip
done

オプションでパスワードレス sudo：

for ip in 192.168.1.{102..103}; do
    ssh -t user@$ip \
        'echo "$(whoami) ALL=(ALL) NOPASSWD: ALL" | sudo tee /etc/sudoers.d/$(whoami)'
done

5. 全ノードへ R をインストール

for host in user@192.168.1.{101..103}; do
    ssh -t $host 'sudo apt update && sudo apt install -y r-base r-base-dev'
done

6. シミュレーション用テンプレート R スクリプト

library(future)
library(future.apply)
library(dplyr)
library(SuperLearner)
library(ranger)
library(xgboost)
library(glmnet)

plan(multicore, workers = 4)   # 各ノードで 4 コア使用

set.seed(1)

# データ生成 ----------------------------------------------------------
n <- 10000
W1 <- rnorm(n); W2 <- rnorm(n)
W3 <- rbinom(n,1,.5); W4 <- rnorm(n)

A  <- rbinom(n,1,plogis(-0.5 + 0.8*W1 + .5*W2^2 + .3*W3 - .4*W1*W2 + .2*W4))
Y  <- rbinom(n,1,plogis(-1 + .2*A + .6*W1 - .4*W2^2 + .5*W3 + .3*W1*W3 + .2*W4^2))

# 真の ATE ------------------------------------------------------------
logit_Y1 <- -1 + .2 + .6*W1 - .4*W2^2 + .5*W3 + .3*W1*W3 + .2*W4^2
logit_Y0 <- -1 + 0    + .6*W1 - .4*W2^2 + .5*W3 + .3*W1*W3 + .2*W4^2

Y1_true <- plogis(logit_Y1)
Y0_true <- plogis(logit_Y0)
true_ATE <- mean(Y1_true - Y0_true)

df <- tibble(W1,W2,W3,W4,A,Y)

# SuperLearner ライブラリ -----------------------------------------------
SL_library <- list(
  c("SL.xgboost","SL.ranger","SL.glm","SL.mean"),
  c("SL.xgboost","SL.ranger"),
  c("SL.xgboost","SL.glm")
)

# TMLE 反復 ---------------------------------------------------------
run_tmle_iteration <- function(seed, df, n_i, SL_lib) {
  set.seed(seed)
  data <- slice_sample(df, n = n_i, replace = TRUE) %>% select(Y,A,W1:W4)
  
  X_outcome   <- data %>% select(A,W1:W4)
  X_treatment <- data %>% select(W1:W4)
  Y_vec       <- data$Y
  A_vec       <- data$A
  
  # Outcome モデル ----------------------------------------------------
  SL_outcome <- SuperLearner(Y = Y_vec, X = X_outcome,
                             family = binomial(), SL.library = SL_lib,
                             cvControl = list(V=5))
  
  outcome      <- predict(SL_outcome, newdata = X_outcome)$pred
  outcome_1    <- predict(SL_outcome, newdata = mutate(X_outcome, A=1))$pred
  outcome_0    <- predict(SL_outcome, newdata = mutate(X_outcome, A=0))$pred
  
  # 予測の境界化 ----------------------------------------------------
  bound <- function(p) pmax(pmin(p,.9999),.0001)
  outcome   <- bound(outcome);   outcome_1 <- bound(outcome_1)
  outcome_0 <- bound(outcome_0)
  
  # Treatment モデル -------------------------------------------------
  SL_treatment <- SuperLearner(Y = A_vec, X = X_treatment,
                               family = binomial(), SL.library = SL_lib,
                               cvControl = list(V=5))
  ps   <- predict(SL_treatment, newdata = X_treatment)$pred
  ps_f <- bound(ps)
  
  # Clever covariate ---------------------------------------------------
  cc  <- ifelse(A_vec==1, 1/ps_f, -1/(1-ps_f))
  
  # Epsilon ------------------------------------------------------------
  eps_model <- glm(Y_vec ~ -1 + offset(qlogis(outcome)) + cc,
                   family = binomial())
  epsilon   <- coef(eps_model)
  
  upd_1 <- plogis(qlogis(outcome_1) + epsilon * (1/ps_f))
  upd_0 <- plogis(qlogis(outcome_0) + epsilon * (-1/(1-ps_f)))
  
  ate  <- mean(upd_1 - upd_0)
  se   <- sqrt(var((Y_vec - ifelse(A_vec==1,upd_1,upd_0))*cc +
                (upd_1-upd_0)-ate) / n_i)
  list(ate=ate,se=se)
}

# 並列実行 -----------------------------------------------------------
n_samples <- 6000
start_end <- 1:1000   # 例：範囲

results <- future_lapply(start_end, function(i){
  run_tmle_iteration(i, df, n_samples, SL_library[[1]])   # 必要に応じてライブラリ変更
})

# 集計 ---------------------------------------------------------------
ate_vec  <- sapply(results, `[[`, "ate")
se_vec   <- sapply(results, `[[`, "se")

summary_tbl <- tibble(iter=start_end, ate=ate_vec, se=se_vec,
                      ci_lower = ate_vec-1.96*se_vec,
                      ci_upper = ate_vec+1.96*se_vec)

# 保存 ---------------------------------------------------------------
dir.create("tmle_results", showWarnings=FALSE)
write_csv(summary_tbl, "tmle_results/summary.csv")

Tip:

SL_library[[1]]

を必要に応じて変更し、ノードの負荷に合わせます。

7. ノードへ作業を分散

# 各ノード用の範囲を決定
total_loop <- 1000
clust_num  <- 3
div_iter   <- total_loop / clust_num
first      <- 1
last       <- div_iter

ranges <- lapply(1:clust_num, function(i){
  if (i == clust_num) paste0(first,":",total_loop)
  else {
    r <- paste0(first,":",round(last))
    first <<- round(first+div_iter)
    last  <<- round(last+div_iter)
    r
  }
})

# プレースホルダを置き換えて SCP
nodes <- c("user@192.168.1.101","user@192.168.1.102","user@192.168.1.103")

for (i in seq_along(nodes)){
  sys_cmd <- sprintf(
    "sed 's/START:END/%s/g' par_test_script.R > par_test_script1.R && \
     scp par_test_script1.R %s:/home/user/par_test_script1.R",
    ranges[[i]], nodes[i])
  system(sys_cmd)
}

8. tmux セッションで Rscript を実行

cluster_name <- "test"

for (node in nodes){
  # 既存セッションを終了（無ければエラーは無視）
  system(sprintf("ssh %s 'tmux kill-session -t %s 2>/dev/null || true'", node, cluster_name))
  
  # デタッチドで新規セッション開始
  system(sprintf("ssh %s 'tmux new-session -d -s %s'", node, cluster_name))
  
  # tmux 内でスクリプト実行し、出力をファイルへリダイレクト
  system(sprintf(
    "ssh %s 'tmux send-keys -t %s \"Rscript par_test_script1.R > result.txt\" ENTER'",
    node, cluster_name))
}

9. 結果収集

library(readr)
df_iter <- tibble()

for (node in nodes){
  for (i in 1:3){   # 各ノードの反復数
    scp_cmd <- sprintf("scp %s:tmle_results/summary.csv summary_%d.csv", node, i)
    system(scp_cmd, intern=TRUE)
    
    tmp <- read_csv(sprintf("summary_%d.csv"))
    df_iter <- bind_rows(df_iter, tmp %>% mutate(node=i))
  }
}

10. 時間と性能の比較

(詳細表は省略)

バイアス & 分散 (CV5 vs CV10)

カバレッジ

主な結論

• CV フォールド数を増やすとバイアスはわずかに減少しますが、分散は増加します。
• 調整済み xgboost + glmnet は低いバイアスと適度なカバレッジを示します。
• GAM + LR は高いカバレッジを持つものの、非対称な尾部が大きく推定量としては不十分です。

改善の余地

• ワークフローを R パッケージ化（SSH・tmux 管理関数、結果集計）
• マスターノードでインストール手順を並列化し、プロビジョニング速度向上
• ノードがバッチ完了したらメール/Slack 通知を送信
• 残り時間の自動推定と未完了反復の再配分実装
• より大規模なクラスターには OpenMPI や Spark の検討

学んだこと

1. sprintf

   +

   system()

   はリモートコマンド実行に強力。

2. future.seed = 100

   を設定すると並列ランダム抽出が再現性を保つ。
3. パッケージインストールパイプラインはノード間で効率的にスクリプト化できる。

4. set.seed()

   は

   future_lapply

   内部で呼び出すと反復ごとの乱数が安定。
5. CV フォールドの選択はバイアス–分散トレードオフを観察しながら決定するべき。

ご質問やコラボレーションのご希望があれば、遠慮なく以下からどうぞ

• GitHub | Twitter | Mastodon | BlueSky

楽しい計算ライフを！