Replies: 4 comments 5 replies
-
|
That's a strange error for a local controller. What do the log files in My first recommendation is to update |
Beta Was this translation helpful? Give feedback.
-
|
Thanks I'll give that a go today and reply. I'm new to crew and I really do love using it, so thanks heaps for the mahi |
Beta Was this translation helpful? Give feedback.
-
|
this is what one of the logs say `R version 4.3.1 (2023-06-16) -- "Beagle Scouts" R is free software and comes with ABSOLUTELY NO WARRANTY. R is a collaborative project with many contributors. Type 'demo()' for some demos, 'help()' for on-line help, or
|
Beta Was this translation helpful? Give feedback.
-
|
I managed to reproduce the error with a simply pipeline. and then tried to push the job manually to crew. I didn't succeed here, mainly because I wasn't sure how to do this. That attempt is here: controller <- crew::crew_controller_local(
name = "snapping_test",
workers = 1,
launch_max = 5L,
local_log_directory = '/tmp/crew',
seconds_idle = 3,
garbage_collection = TRUE
)
controller$start()
tar_load(summariseTripTuples)
controller$push(packages = c('tidyverse'),
name = "summariseTripTuples", command = {
summariseTripTuples %>%
group_by(node_id1, node_id2, weight_category) %>%
summarise(vehicle_count = 1,
trips_n = sum(trips_n, na.rm = T)) %>%
ungroup()
})
monitor <- crew_monitor_local()
monitor$dispatchers()
controller$wait(mode = "all")
controller$summary()
controller$launcher$summary()
controller$client$summary()
task <- controller$pop() # Worker started, task complete.
task
task$result[[1]]
task$error
controller$terminate()
crew_clean()
I didn't know how to pass in the dataframe summariseTripTuples. But because I had loaded it, I saw it was 371million rows and during the loading my RAM hit 57Gb. So I figured I was just asking it too much. Instead I am now reading in each of the 1000 targets in batches of 10 and appending them to a running grouping. That's levelling out at about 2.1million rows. That code is here... I'll write it into the pipeline shortly. target_names <- tar_objects()
target_names <- tarobjectslist[str_detect(target_names,'summariseTripTuples')]
# Define batch size
batch_size <- 10
# Initial setup for the final result
result_summariseTripTuples <- NULL
# Function to process and combine data frames
process_combine <- function(data_frames) {
bind_rows(data_frames) %>%
group_by(node_id1, node_id2, weight_category) %>%
summarise(vehicle_count = 1,
trips_n = sum(trips_n, na.rm = TRUE),
.groups = 'drop')
}
# Split target names into batches
target_batches <- split(target_names, ceiling(seq_along(target_names) / batch_size))
# Iterate over each batch
for (batch in target_batches) {
# Load all targets in the current batch
batch_data <- lapply(batch, function(target_name) {
filelocation <- gsub(' ', '', paste("_targets/objects/", target_name))
qs::qread(filelocation)
})
# Combine all data frames in the current batch
batch_combined <- process_combine(batch_data)
# Combine with the result from previous batches
if (is.null(result_summariseTripTuples)) {
result_summariseTripTuples <- batch_combined
} else {
result_summariseTripTuples <- process_combine(list(result_summariseTripTuples, batch_combined))
}
print(nrow(result_summariseTripTuples))
rm(batch_data, batch_combined)
gc()
}and that failed in the pipeline too |
Beta Was this translation helpful? Give feedback.
-
|
OK so I did this and pushed to one worker and it worked. Still don't know why it doesn't work within a targets pipeline. I updated all the packages. collateNodeCounts <- function(summariseTripTuples){
target_names <- tar_objects()
target_names <- target_names[str_detect(target_names,'summariseTripTuples')]
batch_size <- 10
result_summariseTripTuples <- NULL
# Function to process and combine data frames
process_combine <- function(data_frames) {
bind_rows(data_frames) %>%
group_by(node_id1, node_id2, weight_category) %>%
summarise(vehicle_count = 1,
trips_n = sum(trips_n, na.rm = TRUE),
.groups = 'drop')
}
# Split target names into batches
target_batches <- split(target_names, ceiling(seq_along(target_names) / batch_size))
# Iterate over each batch
for (batch in target_batches) {
# Load all targets in the current batch
batch_data <- lapply(batch, function(target_name) {
filelocation <- gsub(' ', '', paste("_targets/objects/", target_name))
qs::qread(filelocation)
})
# Combine all data frames in the current batch
batch_combined <- process_combine(batch_data)
# Combine with the result from previous batches
if (is.null(result_summariseTripTuples)) {
result_summariseTripTuples <- batch_combined
} else {
result_summariseTripTuples <- process_combine(list(result_summariseTripTuples, batch_combined))
}
print(nrow(result_summariseTripTuples))
rm(batch_data, batch_combined)
gc()
}
return(result_summariseTripTuples)
}
controller <- crew::crew_controller_local(
name = "snapping_test",
workers = 1,
launch_max = 5L,
local_log_directory = '/tmp/crew',
seconds_idle = 3,
garbage_collection = TRUE
)
controller$start()
controller$push(packages = c('tidyverse', 'targets'),
name = "summariseTripTuples", command = {
collateNodeCounts()
})
controller$wait(mode = 'all')
task <- controller$pop()
controller$terminate()
crew_clean()
taskThe number of rows slowly increase towards the unique combinates of node1, node2, and weight category. This is better than the 371M rows!
I got a data frame with the expected number of results. A tibble: 1 × 12
name command result seconds seed algorithm error trace warnings launcher worker instance
<chr> <chr> <list> <dbl> <int> <chr> <chr> <chr> <chr> <chr> <int> <chr>
1 summariseTripTuples NA <tibble [2,295,481 × 5]> 1557. NA NA NA NA NA snapping_test 1 7edd89e3… |
Beta Was this translation helpful? Give feedback.
-
|
The default behavior of You can also use file targets for big data cases like this, in which case your I am not aware that it is possible to do a "partial gather" in |
Beta Was this translation helpful? Give feedback.
-
|
Thanks I was just in the middle of implementing saving ...
tar_target(summariseTripTuples, summariseTripsToNodeTuples(snapped, trip_attributes), map(snapped)),
tar_target(collatedNodeCounts, collateNodeCounts(summariseTripTuples)),
...meant that So are you saying I can do ...
tar_target(summariseTripTuples, summariseTripsToNodeTuples(snapped, trip_attributes), map(snapped)),
tar_target(collatedNodeCounts, collateNodeCounts(summariseTripTuples, retrieval = "none")),
...To maintain the dependency but not load |
Beta Was this translation helpful? Give feedback.
-
|
It sounds like you got it working, but to be clear for anyone in the future which stumbles on this, |
Beta Was this translation helpful? Give feedback.
-
|
Pipeline ran successfully. Thanks for all the help. |
Beta Was this translation helpful? Give feedback.
-
My pipeline does the following:
Last error message: {crew} worker 1 launched 5 times in a row without completing any tasks. Either troubleshoot or raise launch_max above 5.The error message suggests that the memory requirements can be altered, but I don't see how. I'm running this on an AWS EC2 c5a.8xlarge with 32vCPUs and 64Gb RAM.
this is the controller Im using
controller <- crew::crew_controller_local( name = "snapping", workers = 1, launch_max = 10L, local_log_directory = '/tmp/crew', garbage_collection = TRUE )any help to overcome this would be appreciated
Beta Was this translation helpful? Give feedback.
All reactions