Skip to content

R Big Data Processing Best Practices

Objective: Master senior-level R big data processing patterns for enterprise systems. When you need to process massive datasets efficiently, when you want to leverage distributed computing, when you need enterprise-grade big data workflows—these best practices become your weapon of choice.

Core Principles

  • Distributed Computing: Leverage multiple cores and machines
  • Memory Efficiency: Process data larger than available memory
  • Scalability: Design for horizontal and vertical scaling
  • Fault Tolerance: Handle failures gracefully
  • Performance: Optimize for speed and throughput

Apache Spark Integration

SparkR Setup

# R/01-spark-integration.R

#' Create SparkR integration
#'
#' @param spark_config Spark configuration
#' @return SparkR integration
create_sparkr_integration <- function(spark_config) {
  integration <- list(
    spark_context = create_spark_context(spark_config),
    spark_dataframe = create_spark_dataframe(spark_config),
    spark_operations = create_spark_operations(spark_config)
  )

  return(integration)
}

#' Create Spark context
#'
#' @param spark_config Spark configuration
#' @return Spark context
create_spark_context <- function(spark_config) {
  spark_context <- c(
    "# SparkR Integration",
    "library(SparkR)",
    "library(DBI)",
    "",
    "# Initialize Spark",
    "sparkR.session(",
    "  appName = \"R Big Data Processing\",",
    "  master = \"local[*]\",",
    "  sparkConfig = list(",
    "    \"spark.sql.adaptive.enabled\" = \"true\",",
    "    \"spark.sql.adaptive.coalescePartitions.enabled\" = \"true\",",
    "    \"spark.sql.adaptive.skewJoin.enabled\" = \"true\",",
    "    \"spark.sql.adaptive.localShuffleReader.enabled\" = \"true\"",
    "  )",
    ")",
    "",
    "# Create Spark context",
    "sc <- sparkR.session()",
    "",
    "# Get Spark context",
    "spark_context <- sparkR.callJMethod(sc, \"sc\")",
    "",
    "# Set log level",
    "sparkR.callJMethod(spark_context, \"setLogLevel\", \"WARN\")"
  )

  return(spark_context)
}

#' Create Spark DataFrame
#'
#' @param spark_config Spark configuration
#' @return Spark DataFrame
create_spark_dataframe <- function(spark_config) {
  spark_dataframe <- c(
    "# Create Spark DataFrame",
    "library(SparkR)",
    "",
    "# Create DataFrame from R data",
    "create_spark_df <- function(data) {",
    "  # Convert to Spark DataFrame",
    "  spark_df <- createDataFrame(data)",
    "  ",
    "  # Cache for performance",
    "  cache(spark_df)",
    "  ",
    "  return(spark_df)",
    "}",
    "",
    "# Create DataFrame from file",
    "create_spark_df_from_file <- function(file_path, format = \"csv\") {",
    "  if (format == \"csv\") {",
    "    spark_df <- read.df(file_path, source = \"csv\", header = \"true\", inferSchema = \"true\")",
    "  } else if (format == \"parquet\") {",
    "    spark_df <- read.df(file_path, source = \"parquet\")",
    "  } else if (format == \"json\") {",
    "    spark_df <- read.df(file_path, source = \"json\")",
    "  }",
    "  ",
    "  # Cache for performance",
    "  cache(spark_df)",
    "  ",
    "  return(spark_df)",
    "}",
    "",
    "# Create Spark DataFrame",
    "spark_df <- create_spark_df(data)",
    "spark_df_file <- create_spark_df_from_file(\"data.parquet\", format = \"parquet\")"
  )

  return(spark_dataframe)
}

#' Create Spark operations
#'
#' @param spark_config Spark configuration
#' @return Spark operations
create_spark_operations <- function(spark_config) {
  spark_operations <- c(
    "# Spark Operations",
    "library(SparkR)",
    "",
    "# Transform Spark DataFrame",
    "transform_spark_df <- function(spark_df, transformations) {",
    "  for (transformation in transformations) {",
    "    spark_df <- transformation(spark_df)",
    "  }",
    "  ",
    "  return(spark_df)",
    "}",
    "",
    "# Filter data",
    "filter_data <- function(spark_df, condition) {",
    "  filtered_df <- filter(spark_df, condition)",
    "  return(filtered_df)",
    "}",
    "",
    "# Select columns",
    "select_columns <- function(spark_df, columns) {",
    "  selected_df <- select(spark_df, columns)",
    "  return(selected_df)",
    "}",
    "",
    "# Group and aggregate",
    "group_aggregate <- function(spark_df, group_cols, agg_functions) {",
    "  grouped_df <- groupBy(spark_df, group_cols)",
    "  aggregated_df <- agg(grouped_df, agg_functions)",
    "  return(aggregated_df)",
    "}",
    "",
    "# Join DataFrames",
    "join_dataframes <- function(df1, df2, join_cols, join_type = \"inner\") {",
    "  joined_df <- join(df1, df2, df1[[join_cols[1]]] == df2[[join_cols[2]]], join_type)",
    "  return(joined_df)",
    "}",
    "",
    "# Use Spark operations",
    "filtered_df <- filter_data(spark_df, \"age > 18\")",
    "selected_df <- select_columns(filtered_df, c(\"id\", \"name\", \"age\"))",
    "grouped_df <- group_aggregate(selected_df, c(\"category\"), list(count = \"count\", mean_age = \"avg(age)\"))"
  )

  return(spark_operations)
}

Spark SQL

# R/01-spark-integration.R (continued)

#' Create Spark SQL
#'
#' @param sql_config SQL configuration
#' @return Spark SQL
create_spark_sql <- function(sql_config) {
  spark_sql <- list(
    sql_queries = create_sql_queries(sql_config),
    sql_optimization = create_sql_optimization(sql_config),
    sql_udfs = create_sql_udfs(sql_config)
  )

  return(spark_sql)
}

#' Create SQL queries
#'
#' @param sql_config SQL configuration
#' @return SQL queries
create_sql_queries <- function(sql_config) {
  sql_queries <- c(
    "# Spark SQL Queries",
    "library(SparkR)",
    "",
    "# Execute SQL query",
    "execute_sql <- function(query) {",
    "  result <- sql(query)",
    "  return(result)",
    "}",
    "",
    "# Complex SQL query",
    "complex_query <- \"\"",
    "SELECT ",
    "  category,",
    "  COUNT(*) as count,",
    "  AVG(age) as avg_age,",
    "  PERCENTILE_APPROX(age, 0.5) as median_age",
    "FROM users ",
    "WHERE age > 18 ",
    "GROUP BY category ",
    "HAVING COUNT(*) > 100 ",
    "ORDER BY count DESC",
    "\"\"",
    "",
    "# Execute complex query",
    "result <- execute_sql(complex_query)",
    "",
    "# Window functions",
    "window_query <- \"\"",
    "SELECT ",
    "  id,",
    "  name,",
    "  age,",
    "  ROW_NUMBER() OVER (PARTITION BY category ORDER BY age DESC) as rank",
    "FROM users",
    "\"\"",
    "",
    "# Execute window query",
    "window_result <- execute_sql(window_query)"
  )

  return(sql_queries)
}

#' Create SQL optimization
#'
#' @param sql_config SQL configuration
#' @return SQL optimization
create_sql_optimization <- function(sql_config) {
  sql_optimization <- c(
    "# SQL Optimization",
    "library(SparkR)",
    "",
    "# Optimize SQL query",
    "optimize_sql <- function(query) {",
    "  # Enable adaptive query execution",
    "  sql(\"SET spark.sql.adaptive.enabled = true\")",
    "  sql(\"SET spark.sql.adaptive.coalescePartitions.enabled = true\")",
    "  sql(\"SET spark.sql.adaptive.skewJoin.enabled = true\")",
    "  ",
    "  # Execute query",
    "  result <- sql(query)",
    "  ",
    "  return(result)",
    "}",
    "",
    "# Create indexes",
    "create_indexes <- function(table_name, columns) {",
    "  for (column in columns) {",
    "    index_query <- paste(\"CREATE INDEX IF NOT EXISTS idx_\", table_name, \"_\", column, \" ON \", table_name, \"(\", column, \")\")",
    "    sql(index_query)",
    "  }",
    "}",
    "",
    "# Use SQL optimization",
    "optimized_result <- optimize_sql(complex_query)",
    "create_indexes(\"users\", c(\"category\", \"age\"))"
  )

  return(sql_optimization)
}

#' Create SQL UDFs
#'
#' @param sql_config SQL configuration
#' @return SQL UDFs
create_sql_udfs <- function(sql_config) {
  sql_udfs <- c(
    "# SQL User Defined Functions",
    "library(SparkR)",
    "",
    "# Register UDF",
    "register_udf <- function(function_name, r_function, return_type) {",
    "  udf <- createUDF(function_name, r_function, return_type)",
    "  registerUDF(udf)",
    "}",
    "",
    "# Define R functions",
    "calculate_age_group <- function(age) {",
    "  if (age < 18) return(\"child\")",
    "  else if (age < 30) return(\"young\")",
    "  else if (age < 50) return(\"adult\")",
    "  else return(\"senior\")",
    "}",
    "",
    "calculate_bmi <- function(weight, height) {",
    "  return(weight / (height / 100) ^ 2)",
    "}",
    "",
    "# Register UDFs",
    "register_udf(\"age_group\", calculate_age_group, \"string\")",
    "register_udf(\"bmi\", calculate_bmi, \"double\")",
    "",
    "# Use UDFs in SQL",
    "udf_query <- \"\"",
    "SELECT ",
    "  id,",
    "  name,",
    "  age,",
    "  age_group(age) as age_group,",
    "  bmi(weight, height) as bmi",
    "FROM users",
    "\"\"",
    "",
    "# Execute UDF query",
    "udf_result <- sql(udf_query)"
  )

  return(sql_udfs)
}

Apache Arrow Integration

Arrow Data Processing

# R/02-arrow-integration.R

#' Create Arrow integration
#'
#' @param arrow_config Arrow configuration
#' @return Arrow integration
create_arrow_integration <- function(arrow_config) {
  integration <- list(
    arrow_setup = create_arrow_setup(arrow_config),
    arrow_operations = create_arrow_operations(arrow_config),
    arrow_optimization = create_arrow_optimization(arrow_config)
  )

  return(integration)
}

#' Create Arrow setup
#'
#' @param arrow_config Arrow configuration
#' @return Arrow setup
create_arrow_setup <- function(arrow_config) {
  arrow_setup <- c(
    "# Apache Arrow Integration",
    "library(arrow)",
    "library(dplyr)",
    "",
    "# Set Arrow options",
    "set_arrow_options <- function() {",
    "  # Set memory pool",
    "  options(arrow.memory_pool = \"system\")",
    "  ",
    "  # Set compression",
    "  options(arrow.compression = \"lz4\")",
    "  ",
    "  # Set batch size",
    "  options(arrow.batch_size = 10000)",
    "  ",
    "  # Set thread count",
    "  options(arrow.use_threads = TRUE)",
    "  options(arrow.num_threads = parallel::detectCores())",
    "}",
    "",
    "# Initialize Arrow",
    "set_arrow_options()",
    "",
    "# Create Arrow table",
    "create_arrow_table <- function(data) {",
    "  # Convert to Arrow table",
    "  arrow_table <- as_arrow_table(data)",
    "  ",
    "  return(arrow_table)",
    "}",
    "",
    "# Create Arrow table from file",
    "create_arrow_table_from_file <- function(file_path, format = \"parquet\") {",
    "  if (format == \"parquet\") {",
    "    arrow_table <- read_parquet(file_path)",
    "  } else if (format == \"arrow\") {",
    "    arrow_table <- read_arrow(file_path)",
    "  } else if (format == \"csv\") {",
    "    arrow_table <- read_csv_arrow(file_path)",
    "  }",
    "  ",
    "  return(arrow_table)",
    "}",
    "",
    "# Create Arrow tables",
    "arrow_table <- create_arrow_table(data)",
    "arrow_table_file <- create_arrow_table_from_file(\"data.parquet\", format = \"parquet\")"
  )

  return(arrow_setup)
}

#' Create Arrow operations
#'
#' @param arrow_config Arrow configuration
#' @return Arrow operations
create_arrow_operations <- function(arrow_config) {
  arrow_operations <- c(
    "# Arrow Operations",
    "library(arrow)",
    "library(dplyr)",
    "",
    "# Filter Arrow table",
    "filter_arrow_table <- function(arrow_table, condition) {",
    "  filtered_table <- arrow_table %>%",
    "    filter(!!rlang::parse_expr(condition))",
    "  ",
    "  return(filtered_table)",
    "}",
    "",
    "# Select columns from Arrow table",
    "select_arrow_columns <- function(arrow_table, columns) {",
    "  selected_table <- arrow_table %>%",
    "    select(all_of(columns))",
    "  ",
    "  return(selected_table)",
    "}",
    "",
    "# Group and aggregate Arrow table",
    "group_aggregate_arrow <- function(arrow_table, group_cols, agg_functions) {",
    "  grouped_table <- arrow_table %>%",
    "    group_by(across(all_of(group_cols))) %>%",
    "    summarise(across(everything(), agg_functions))",
    "  ",
    "  return(grouped_table)",
    "}",
    "",
    "# Join Arrow tables",
    "join_arrow_tables <- function(table1, table2, join_cols, join_type = \"inner\") {",
    "  joined_table <- table1 %>%",
    "    inner_join(table2, by = join_cols)",
    "  ",
    "  return(joined_table)",
    "}",
    "",
    "# Use Arrow operations",
    "filtered_arrow <- filter_arrow_table(arrow_table, \"age > 18\")",
    "selected_arrow <- select_arrow_columns(filtered_arrow, c(\"id\", \"name\", \"age\"))",
    "grouped_arrow <- group_aggregate_arrow(selected_arrow, c(\"category\"), list(count = \"count\", mean_age = \"mean\"))"
  )

  return(arrow_operations)
}

#' Create Arrow optimization
#'
#' @param arrow_config Arrow configuration
#' @return Arrow optimization
create_arrow_optimization <- function(arrow_config) {
  arrow_optimization <- c(
    "# Arrow Optimization",
    "library(arrow)",
    "library(dplyr)",
    "",
    "# Optimize Arrow table",
    "optimize_arrow_table <- function(arrow_table) {",
    "  # Repartition for better performance",
    "  optimized_table <- arrow_table %>%",
    "    group_by(across(everything())) %>%",
    "    summarise(across(everything(), list)) %>%",
    "    ungroup()",
    "  ",
    "  return(optimized_table)",
    "}",
    "",
    "# Cache Arrow table",
    "cache_arrow_table <- function(arrow_table) {",
    "  # Cache table in memory",
    "  cached_table <- arrow_table %>%",
    "    compute()",
    "  ",
    "  return(cached_table)",
    "}",
    "",
    "# Write optimized Arrow table",
    "write_optimized_arrow <- function(arrow_table, output_path, format = \"parquet\") {",
    "  if (format == \"parquet\") {",
    "    write_parquet(arrow_table, output_path)",
    "  } else if (format == \"arrow\") {",
    "    write_arrow(arrow_table, output_path)",
    "  } else if (format == \"csv\") {",
    "    write_csv_arrow(arrow_table, output_path)",
    "  }",
    "}",
    "",
    "# Use Arrow optimization",
    "optimized_arrow <- optimize_arrow_table(arrow_table)",
    "cached_arrow <- cache_arrow_table(optimized_arrow)",
    "write_optimized_arrow(cached_arrow, \"output.parquet\", format = \"parquet\")"
  )

  return(arrow_optimization)
}

Distributed Computing

Parallel Processing

# R/03-distributed-computing.R

#' Create distributed computing
#'
#' @param distributed_config Distributed configuration
#' @return Distributed computing
create_distributed_computing <- function(distributed_config) {
  distributed <- list(
    parallel_setup = create_parallel_setup(distributed_config),
    distributed_operations = create_distributed_operations(distributed_config),
    fault_tolerance = create_fault_tolerance(distributed_config)
  )

  return(distributed)
}

#' Create parallel setup
#'
#' @param distributed_config Distributed configuration
#' @return Parallel setup
create_parallel_setup <- function(distributed_config) {
  parallel_setup <- c(
    "# Distributed Computing Setup",
    "library(parallel)",
    "library(foreach)",
    "library(doParallel)",
    "library(future)",
    "library(future.apply)",
    "",
    "# Setup parallel backend",
    "setup_parallel_backend <- function(backend = \"multicore\", cores = NULL) {",
    "  if (is.null(cores)) {",
    "    cores <- parallel::detectCores()",
    "  }",
    "  ",
    "  if (backend == \"multicore\") {",
    "    registerDoParallel(cores = cores)",
    "  } else if (backend == \"snow\") {",
    "    cl <- makeCluster(cores)",
    "    registerDoParallel(cl)",
    "  } else if (backend == \"future\") {",
    "    plan(multicore, workers = cores)",
    "  }",
    "  ",
    "  return(cores)",
    "}",
    "",
    "# Setup parallel backend",
    "cores <- setup_parallel_backend(backend = \"multicore\")",
    "",
    "# Create parallel workers",
    "create_parallel_workers <- function(n_workers = NULL) {",
    "  if (is.null(n_workers)) {",
    "    n_workers <- parallel::detectCores()",
    "  }",
    "  ",
    "  workers <- makeCluster(n_workers)",
    "  ",
    "  return(workers)",
    "}",
    "",
    "# Create workers",
    "workers <- create_parallel_workers()"
  )

  return(parallel_setup)
}

#' Create distributed operations
#'
#' @param distributed_config Distributed configuration
#' @return Distributed operations
create_distributed_operations <- function(distributed_config) {
  distributed_operations <- c(
    "# Distributed Operations",
    "library(parallel)",
    "library(foreach)",
    "library(doParallel)",
    "",
    "# Distribute data processing",
    "distribute_data_processing <- function(data, function_name, chunk_size = 1000) {",
    "  # Split data into chunks",
    "  n_chunks <- ceiling(nrow(data) / chunk_size)",
    "  chunks <- split(data, rep(1:n_chunks, each = chunk_size, length.out = nrow(data)))",
    "  ",
    "  # Process chunks in parallel",
    "  results <- foreach(chunk = chunks, .combine = rbind, .packages = c(\"data.table\", \"dplyr\")) %dopar% {",
    "    function_name(chunk)",
    "  }",
    "  ",
    "  return(results)",
    "}",
    "",
    "# Distribute file processing",
    "distribute_file_processing <- function(file_paths, function_name) {",
    "  # Process files in parallel",
    "  results <- foreach(file_path = file_paths, .combine = rbind, .packages = c(\"data.table\", \"readr\")) %dopar% {",
    "    function_name(file_path)",
    "  }",
    "  ",
    "  return(results)",
    "}",
    "",
    "# Distribute aggregation",
    "distribute_aggregation <- function(data, group_cols, agg_functions, chunk_size = 1000) {",
    "  # Split data into chunks",
    "  n_chunks <- ceiling(nrow(data) / chunk_size)",
    "  chunks <- split(data, rep(1:n_chunks, each = chunk_size, length.out = nrow(data)))",
    "  ",
    "  # Process chunks in parallel",
    "  chunk_results <- foreach(chunk = chunks, .combine = list, .packages = c(\"data.table\")) %dopar% {",
    "    chunk_dt <- as.data.table(chunk)",
    "    chunk_dt[, lapply(agg_functions, function(f) f(.SD)), by = group_cols]",
    "  }",
    "  ",
    "  # Combine results",
    "  combined_results <- rbindlist(chunk_results)",
    "  ",
    "  # Final aggregation",
    "  final_results <- combined_results[, lapply(agg_functions, function(f) f(.SD)), by = group_cols]",
    "  ",
    "  return(final_results)",
    "}",
    "",
    "# Use distributed operations",
    "distributed_results <- distribute_data_processing(data, function_name = clean_data)",
    "file_results <- distribute_file_processing(file_paths, function_name = read_file)",
    "aggregated_results <- distribute_aggregation(data, group_cols = c(\"category\"), agg_functions = list(count = length, mean_value = function(x) mean(x$value)))"
  )

  return(distributed_operations)
}

#' Create fault tolerance
#'
#' @param distributed_config Distributed configuration
#' @return Fault tolerance
create_fault_tolerance <- function(distributed_config) {
  fault_tolerance <- c(
    "# Fault Tolerance",
    "library(parallel)",
    "library(foreach)",
    "library(doParallel)",
    "",
    "# Fault-tolerant parallel processing",
    "fault_tolerant_processing <- function(data, function_name, max_retries = 3) {",
    "  # Split data into chunks",
    "  n_chunks <- ceiling(nrow(data) / 1000)",
    "  chunks <- split(data, rep(1:n_chunks, each = 1000, length.out = nrow(data)))",
    "  ",
    "  # Process chunks with retry logic",
    "  results <- foreach(chunk = chunks, .combine = rbind, .packages = c(\"data.table\", \"dplyr\")) %dopar% {",
    "    retry_count <- 0",
    "    success <- FALSE",
    "    ",
    "    while (retry_count < max_retries && !success) {",
    "      tryCatch({",
    "        result <- function_name(chunk)",
    "        success <- TRUE",
    "      }, error = function(e) {",
    "        retry_count <<- retry_count + 1",
    "        if (retry_count >= max_retries) {",
    "          stop(paste(\"Max retries exceeded for chunk:\", e$message))",
    "        }",
    "        Sys.sleep(1)  # Wait before retry",
    "      })",
    "    }",
    "    ",
    "    if (success) result else NULL",
    "  }",
    "  ",
    "  return(results)",
    "}",
    "",
    "# Fault-tolerant file processing",
    "fault_tolerant_file_processing <- function(file_paths, function_name, max_retries = 3) {",
    "  # Process files with retry logic",
    "  results <- foreach(file_path = file_paths, .combine = rbind, .packages = c(\"data.table\", \"readr\")) %dopar% {",
    "    retry_count <- 0",
    "    success <- FALSE",
    "    ",
    "    while (retry_count < max_retries && !success) {",
    "      tryCatch({",
    "        result <- function_name(file_path)",
    "        success <- TRUE",
    "      }, error = function(e) {",
    "        retry_count <<- retry_count + 1",
    "        if (retry_count >= max_retries) {",
    "          stop(paste(\"Max retries exceeded for file\", file_path, \":\", e$message))",
    "        }",
    "        Sys.sleep(1)  # Wait before retry",
    "      })",
    "    }",
    "    ",
    "    if (success) result else NULL",
    "  }",
    "  ",
    "  return(results)",
    "}",
    "",
    "# Use fault-tolerant processing",
    "fault_tolerant_results <- fault_tolerant_processing(data, function_name = clean_data)",
    "fault_tolerant_file_results <- fault_tolerant_file_processing(file_paths, function_name = read_file)"
  )

  return(fault_tolerance)
}

Memory Management

Memory Optimization

# R/04-memory-management.R

#' Create memory optimization
#'
#' @param memory_config Memory configuration
#' @return Memory optimization
create_memory_optimization <- function(memory_config) {
  memory <- list(
    memory_monitoring = create_memory_monitoring(memory_config),
    memory_cleanup = create_memory_cleanup(memory_config),
    memory_efficient_operations = create_memory_efficient_operations(memory_config)
  )

  return(memory)
}

#' Create memory monitoring
#'
#' @param memory_config Memory configuration
#' @return Memory monitoring
create_memory_monitoring <- function(memory_config) {
  memory_monitoring <- c(
    "# Memory Monitoring",
    "library(pryr)",
    "library(profmem)",
    "",
    "# Monitor memory usage",
    "monitor_memory <- function() {",
    "  memory_info <- list(",
    "    total_memory = memory.size(),",
    "    used_memory = memory.size() - memory.size(max = FALSE),",
    "    available_memory = memory.size(max = FALSE),",
    "    gc_memory = gc()",
    "  )",
    "  ",
    "  return(memory_info)",
    "}",
    "",
    "# Monitor object sizes",
    "monitor_object_sizes <- function() {",
    "  objects <- ls(envir = .GlobalEnv)",
    "  sizes <- sapply(objects, function(x) object.size(get(x)))",
    "  sizes <- sort(sizes, decreasing = TRUE)",
    "  ",
    "  return(sizes)",
    "}",
    "",
    "# Monitor memory usage",
    "memory_info <- monitor_memory()",
    "object_sizes <- monitor_object_sizes()",
    "print(memory_info)",
    "print(head(object_sizes))"
  )

  return(memory_monitoring)
}

#' Create memory cleanup
#'
#' @param memory_config Memory configuration
#' @return Memory cleanup
create_memory_cleanup <- function(memory_config) {
  memory_cleanup <- c(
    "# Memory Cleanup",
    "library(pryr)",
    "",
    "# Clean up memory",
    "cleanup_memory <- function() {",
    "  # Force garbage collection",
    "  gc(verbose = TRUE)",
    "  ",
    "  # Clear large objects",
    "  large_objects <- ls(envir = .GlobalEnv)[sapply(ls(envir = .GlobalEnv), function(x) object.size(get(x))) > 100 * 1024 * 1024]",
    "  if (length(large_objects) > 0) {",
    "    rm(list = large_objects, envir = .GlobalEnv)",
    "    gc(verbose = TRUE)",
    "  }",
    "  ",
    "  return(TRUE)",
    "}",
    "",
    "# Clean up memory periodically",
    "cleanup_memory_periodically <- function(interval = 300) {",
    "  while (TRUE) {",
    "    Sys.sleep(interval)",
    "    cleanup_memory()",
    "  }",
    "}",
    "",
    "# Use memory cleanup",
    "cleanup_memory()"
  )

  return(memory_cleanup)
}

#' Create memory efficient operations
#'
#' @param memory_config Memory configuration
#' @return Memory efficient operations
create_memory_efficient_operations <- function(memory_config) {
  memory_efficient_operations <- c(
    "# Memory Efficient Operations",
    "library(data.table)",
    "library(dplyr)",
    "",
    "# Process data in chunks",
    "process_data_in_chunks <- function(data, function_name, chunk_size = 1000) {",
    "  n_chunks <- ceiling(nrow(data) / chunk_size)",
    "  results <- list()",
    "  ",
    "  for (i in 1:n_chunks) {",
    "    start_idx <- (i - 1) * chunk_size + 1",
    "    end_idx <- min(i * chunk_size, nrow(data))",
    "    ",
    "    chunk <- data[start_idx:end_idx, ]",
    "    result <- function_name(chunk)",
    "    results[[i]] <- result",
    "    ",
    "    # Clean up chunk",
    "    rm(chunk)",
    "    gc()",
    "  }",
    "  ",
    "  # Combine results",
    "  combined_results <- rbindlist(results)",
    "  ",
    "  return(combined_results)",
    "}",
    "",
    "# Use memory efficient operations",
    "processed_data <- process_data_in_chunks(data, function_name = clean_data)"
  )

  return(memory_efficient_operations)
}

TL;DR Runbook

Quick Start

# 1. Create SparkR integration
spark_integration <- create_sparkr_integration(spark_config)

# 2. Create Arrow integration
arrow_integration <- create_arrow_integration(arrow_config)

# 3. Create distributed computing
distributed <- create_distributed_computing(distributed_config)

# 4. Create memory optimization
memory <- create_memory_optimization(memory_config)

Essential Patterns

# Complete big data processing pipeline
create_big_data_processing_pipeline <- function(pipeline_config) {
  # Create SparkR integration
  spark_integration <- create_sparkr_integration(pipeline_config$spark_config)

  # Create Arrow integration
  arrow_integration <- create_arrow_integration(pipeline_config$arrow_config)

  # Create distributed computing
  distributed <- create_distributed_computing(pipeline_config$distributed_config)

  # Create memory optimization
  memory <- create_memory_optimization(pipeline_config$memory_config)

  return(list(
    spark = spark_integration,
    arrow = arrow_integration,
    distributed = distributed,
    memory = memory
  ))
}

This guide provides the complete machinery for implementing big data processing for R applications. Each pattern includes implementation examples, optimization strategies, and real-world usage patterns for enterprise big data systems.