Hi Team,
this code ran failed using legacy ray run...
# daft-devc
can cai
11/09/2025, 9:29 AMHi Team,
this code ran failed using legacy ray runner as driver oom, i'm not sure what the specific reason is. May I ask if anyone else has had such an experience?
from daft import contextimport daftdaft.context.set_runner_ray(noop_if_initialized=True)context.set_execution_config(use_legacy_ray_runner=True)df = (daft.read_parquet("/mnt/code/all_github_repos/repo_infos/**/*.parquet").select(daft.col("content")["full_name"].alias("repo_name"),daft.col("content")["stargazers_count"].alias("stars"),).with_column("repo_name_normalized",daft.col("repo_name").str.replace("[./\-]", "_").str.lower(),).filter("stars > 0").select("repo_name_normalized", "repo_name", "stars"))v1v2 = daft.read_parquet("/mnt/code/repo_level_code/1_repo_name_dedup/**/*.parquet")df.join(v1v2, on=df["repo_name_normalized"] == v1v2["repo_name_normalized"], how="anti").write_parquet("/mnt/code/all_github_repos/repos_sub_v1v2_sortby_star_tmp22/")Austin TUI Memory Profile CPU 146% ████████ MEM4658M ████████ 1/2_________ Command python /tmp/test/debug.py⎝__⎠ ⎝__⎠ Python 3.10.15 PID 710426 PID:TID 710426:0:710816Samples 10647 :timer_clock: 26" Threshold 0%OWN TOTAL %OWN %TOTAL FUNCTION0B 3604M 0.0% 77.4% _bootstrap (/home/ray/anaconda3/lib/python3.10/threading.py:973) │0B 3604M 0.0% 77.4% _bootstrap_inner (/home/ray/anaconda3/lib/python3.10/threading.py:1016) │0B 3604M 0.0% 77.4% run (/home/ray/anaconda3/lib/python3.10/threading.py:953) │3M 2352M 0.1% 50.5% _run_plan (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1451) │0B 2348M 0.0% 50.4% _dispatch_tasks (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1294) │5M 1983M 0.1% 42.6% _build_partitions (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1567) │0B 1978M 0.0% 42.5% remote (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_tracing.py:627) │160K 1978M 0.0% 42.5% remote (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/remote_function.py:293) │0B 1978M 0.0% 42.5% auto_init_wrapper (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/_private/auto_init_hook.py:22) │264K 1978M 0.0% 42.5% _invocation_remote_span (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/util/tracing/tracing_helper.py:31│0B 1977M 0.0% 42.5% _remote (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/remote_function.py:525) │1657M 1977M 35.6% 42.5% invocation (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/remote_function.py:490) │200K 319M 0.0% 6.9% serialize (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/_private/serialization.py:672) │624K 293M 0.0% 6.3% _serialize_to_msgpack (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/_private/serialization.py:605) │1M 291M 0.0% 6.3% _serialize_to_pickle5 (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/_private/serialization.py:547) │0B 289M 0.0% 6.2% dumps (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/cloudpickle/cloudpickle.py:1479) │249M 288M 5.4% 6.2% dump (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/cloudpickle/cloudpickle.py:1245) │Austin TUI Memory Profile CPU 118% ████████ MEM5716M ████████ 1/2_________ Command python /tmp/test/debug.py⎝__⎠ ⎝__⎠ Python 3.10.15 PID 710426 PID:TID 710426:0:710816Samples 16954 :timer_clock: 37" Threshold 0%OWN TOTAL %OWN %TOTAL FUNCTION0B 5880M 0.0% 100.0% _bootstrap (/home/ray/anaconda3/lib/python3.10/threading.py:973) │0B 5880M 0.0% 100.0% _bootstrap_inner (/home/ray/anaconda3/lib/python3.10/threading.py:1016) │0B 5880M 0.0% 100.0% run (/home/ray/anaconda3/lib/python3.10/threading.py:953) │348K 1101M 0.0% 18.7% _run_plan (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1476) │0B 700M 0.0% 11.9% _await_tasks (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1319) │0B 700M 0.0% 11.9% auto_init_wrapper (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/_private/auto_init_hook.py:22) │0B 700M 0.0% 11.9% wrapper (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/_private/client_mode_hook.py:104) │501M 501M 8.5% 8.5% wait (/root/micromamba/envs/user_env/lib/python3.10/site-packages/ray/_private/worker.py:3113) │Austin TUI Memory Profile CPU 438% ████████ MEM7293M ████████ 1/3_________ Command python /tmp/test/debug.py⎝__⎠ ⎝__⎠ Python 3.10.15 PID 710426 PID:TID 710426:0:710816Samples 27223 :timer_clock: 55" Threshold 0%OWN TOTAL %OWN %TOTAL FUNCTION0B 8113M 0.0% 100.0% _bootstrap (/home/ray/anaconda3/lib/python3.10/threading.py:973) │0B 8113M 0.0% 100.0% _bootstrap_inner (/home/ray/anaconda3/lib/python3.10/threading.py:1016) │0B 8113M 0.0% 100.0% run (/home/ray/anaconda3/lib/python3.10/threading.py:953) │0B 676M 0.0% 9.3% _run_plan (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1434) │212K 676M 0.0% 9.3% _construct_dispatch_batch (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1236) │0B 674M 0.0% 9.3% __next__ (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_tracing.py:639) │40K 673M 0.0% 9.2% __iter__ (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:1899) │0B 673M 0.0% 9.2% file_write (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:134) │0B 673M 0.0% 9.2% <genexpr> (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:134) │0B 673M 0.0% 9.2% pipeline_instruction (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:243) │0B 673M 0.0% 9.2% <genexpr> (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:243) │0B 673M 0.0% 9.2% hash_join (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:397) │0B 673M 0.0% 9.2% reduce (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:1616) │0B 673M 0.0% 9.2% pipeline_instruction (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:243) │264K 663M 0.0% 9.1% <genexpr> (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/physical_plan.py:244) │324K 663M 0.0% 9.1% add_instruction (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/execution_step.py:146) │0B 662M 0.0% 9.1% run_partial_metadata (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/execution_step.py:1124) │302M 631M 4.1% 8.7% <listcomp> (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/execution_step.py:1125) │Austin TUI Memory Profile CPU 110% ███▇████ MEM1..1M ████████ 1/3_________ Command python /tmp/test/debug.py⎝__⎠ ⎝__⎠ Python 3.10.15 PID 710426 PID:TID 710426:0:710816Samples 74891 :timer_clock: 2'08" Threshold 0%OWN TOTAL %OWN %TOTAL FUNCTION0B 19478M 0.0% 100.0% _bootstrap (/home/ray/anaconda3/lib/python3.10/threading.py:973) │0B 19478M 0.0% 100.0% _bootstrap_inner (/home/ray/anaconda3/lib/python3.10/threading.py:1016) │0B 19478M 0.0% 100.0% run (/home/ray/anaconda3/lib/python3.10/threading.py:953) │248K 854M 0.0% 5.2% _run_plan (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/ray_runner.py:1487) │0B 853M 0.0% 5.2% set_done (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/execution_step.py:80) │0B 852M 0.0% 5.2% cache_metadata (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/execution_step.py:301) │7M 849M 0.0% 5.2% <listcomp> (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/execution/execution_step.py:302) │9M 19M 0.1% 0.1% merge_with_partial (/root/micromamba/envs/user_env/lib/python3.10/site-packages/daft/runners/partitioning.py:100) │6M 6M 0.0% 0.0% __init__ (<string>:5) │can cai
11/09/2025, 3:53 PM Copy code
ObjectType ObjectCount ObjectSize(MB) ObjectAvgSize(B)
----------------------------------------------------------------------
PartitionMetadata 1,742,158 79.75 48.0
dict 146,829 29.99 214.2
deque 293 27.01 96646.1
list 105,195 10.77 107.4
function 52,905 7.27 144.0
type 6,373 6.45 1061.3
tuple 35,128 2.07 61.8
PartialPartitionMetadata 31,111 1.42 48.0
RayMaterializedResult 31,111 1.42 48.0
set 2,122 1.08 531.5
ReferenceType 10,552 0.72 72.0
PartitionMetadataAccessor 15,690 0.72 48.0
SingleOutputPartitionTask 15,689 0.72 48.0
ScanWithTask 15,423 0.71 48.0
MicroPartition 15,423 0.71 48.0
ReduceMergeAndSort 15,422 0.71 48.0
cython_function_or_method 3,017 0.60 208.0
ABCMeta 575 0.58 1060.8
Name 12,137 0.56 48.0
defaultdict 25 0.54 22748.2
GeneratedProtocolMessageType 566 0.49 904.0
getset_descriptor 7,799 0.48 64.0
frozenset 1,368 0.46 351.7
method 5,866 0.36 64.0
cell 8,899 0.34 40.0
Pattern 493 0.30 644.8
wrapper_descriptor 3,995 0.27 72.0
method_descriptor 3,861 0.27 72.0
builtin_function_or_method 3,560 0.24 72.0
OrderedDict 436 0.21 513.5
MetaHasTraits 217 0.21 1021.5
Attribute 3,890 0.19 52.0
property 2,483 0.19 80.0
module 2,692 0.18 72.0
member_descriptor 2,919 0.18 64.0
EnumMeta 159 0.16 1066.7
Constant 3,283 0.15 48.0
Call 3,075 0.14 48.0
ModuleSpec 2,682 0.12 48.0
_ProtocolMeta 123 0.12 1046.0
SourceFileLoader 2,503 0.11 48.0
ForwardRef 1,199 0.11 96.0
fused_cython_function 493 0.11 224.0
Parameter 1,713 0.10 64.0
_TypedDictMeta 100 0.10 1072.0
Field 880 0.10 119.9
arg 1,617 0.07 48.0
Assign 1,506 0.07 48.0
classmethod 1,439 0.07 48.0
LazySet 2 0.06 32992.0can cai
11/10/2025, 2:04 AMgently ping @Colin Ho
May I ask if you can offer me some help？
c
Colin Ho
11/10/2025, 6:54 AMThe objects themselves are small but each objectref takes up 3kb https://github.com/ray-project/ray/blob/
c
can cai
11/10/2025, 9:09 AM@Colin Ho Thanks for your reply.
I'm very sorry, but this url returns a 404.
c
Colin Ho
11/10/2025, 9:09 AMc
can cai
11/10/2025, 9:12 AMThank you very much. @Colin Ho
But I'm curious why I can't see the memory size occupied by these object references in the python heap?
Are they using off-heap memory?
c
Colin Ho
11/10/2025, 12:42 PMYeah ray uses c++ under the hood, so your profiler might be missing a lot of the allocations from there
🙏 1
c
can cai
11/14/2025, 8:04 AMHi @Colin Ho
It doesn't seem to be caused by this reason.
I reduced the value of
CALLER_MEMORY_USAGE_PER_OBJECT_REFc
Colin Ho
11/14/2025, 8:10 AMHow did u change it?
Colin Ho
11/14/2025, 8:11 AMDid you just change the value of the constant?
c
can cai
11/14/2025, 8:23 AM@Colin Ho yes, i just change the value of the constant in ray_constants.py.
c
Colin Ho
11/14/2025, 8:24 AMYeah i don't think that's gonna work. That constant is just used as an estimate for memory usage in their scheduler https://github.com/search?q=repo%3Aray-project%2Fray+%22CALLER_MEMORY_USAGE_PER_OBJECT_REF%22&type=code , it doesn't actually change the real memory usage
c
can cai
11/14/2025, 8:29 AMI think so too.
May I ask if there are any other reasons? I couldn't run join on flotilla runner, because the shuffle phase would cause worker oom killed.
Therefor, I want to run it on legacy ray runner first, but driver oom killed is another serious problem.
c
Colin Ho
11/14/2025, 8:30 AMFor joins i would suggest doing a manual repartition (for both sides) before the join, that way you can control the number of partitions for the shuffle
Colin Ho
11/14/2025, 8:31 AMby default the number of shuffle partitions is the larger side. For example if one side is 100 partitions and the other side is 200 partitions, we will do 200 partitions for the shuffle. But if you add a manual
df.repartition(150, "join_key)Colin Ho
11/14/2025, 8:32 AMWe're actively working on better algorithms to determine the shuffle partitions, but in the meantime this should allow you to test
c
can cai
11/14/2025, 8:39 AMIf there are more micropartitions, is it more likely to cause legacy ray runner oom?
my dataset is very large. The left side has over 100,000 partitions. I tried jumping to a larger morsel to reduce Micropartitions, but it still couldn't reduce the usage of driver memory.
can cai
11/14/2025, 8:42 AMOn the contrary, it is more likely to cause worker oom killed. 😂
c
Colin Ho
11/14/2025, 8:51 AMsmaller / more partitions can prevent worker OOM, but increase risk of driver OOM.
On the otherhand, larger / less partitions can increase risk of worker OOM but not driver.
The sweet spot is probably somewhere in the middle, and it depends on the cluster resources as well.
We're working on a shuffle that doesn't use ray object refs, and instead writes directly to disk, which should help. But this is a pretty big project and not on our short term priority, likely it will happen next year.
c
can cai
11/14/2025, 8:57 AMi got it. thanks.
Is there any better short-term solution that can help me run it through? Whether it's on flotilla or legacy runner.
c
Colin Ho
11/14/2025, 11:38 PMBest best for now is to do the repartition before the join to control the partitions
Colin Ho
11/14/2025, 11:38 PMAnd also try this on flotilla as well if possible
c
can cai
11/16/2025, 10:53 AMThank you. @Colin Ho i will try to repartition before the join to control the partitions.
I initially executed it on flotilla, but it was very easy to cause the worker to fail for oom killed, and the flotilla actor would also experience continuous memory increase issues, resulting in both the driver and the worker needing to be configured with a large amount of memory.
can cai
11/16/2025, 10:56 AMLike this, I still haven't been able to run a join of a large dataset successfully so far, whether on legacy or flotilla runner.
can cai
11/17/2025, 3:34 AMfrom daft import contextimport daftdaft.context.set_runner_ray(noop_if_initialized=True)df = (daft.read_parquet("/mnt/test/**/*.parquet").repartition(5000).select(daft.col("content")["full_name"].alias("repo_name"),daft.col("content")["stargazers_count"].alias("stars"),).with_column("repo_name_normalized",daft.col("repo_name").str.replace("[./\-]", "_").str.lower(),).filter("stars > 0").select("repo_name_normalized", "repo_name", "stars"))v1v2 = daft.read_parquet("/mnt/test/1_repo_name_dedup/**/*.parquet").repartition(5000)df.join(v1v2, on=df["repo_name_normalized"] == v1v2["repo_name_normalized"], how="anti").count_rows() Copy code
(pid=24974) PhysicalScan 0: 0%| | 0.00/1.00 [00:00<?, ?it/s]
(pid=24974) PhysicalScan 1: 0%| | 0.00/1.00 [00:00<?, ?it/s]
(pid=24974) InMemoryScan->Repartition 2: 0%| | 0.00/1.00 [00:00<?, ?it/s]
(pid=24974) InMemoryScan 3: 0%| | 0.00/1.00 [00:00<?, ?it/s]can cai
11/17/2025, 3:35 AMcc @jay @Colin Ho
c
Colin Ho
11/17/2025, 1:15 PMdo you have a copy of the plan?
df.explain(True)c
can cai
11/18/2025, 11:47 AMyes, here is the physical plan. cc @Colin Ho
== Physical Plan ==* Project: col(2: repo_name_normalized), col(0: repo_name), col(1: stars)| Resource request = None|* Hash Join| Left on: col(2: repo_name_normalized) == col(2: repo_name_normalized)| Right on: col(0: repo_name_normalized) == col(0: repo_name_normalized)|\| * Repartition: Hash| | Num partitions = Some(5000)| | By = col(0: repo_name_normalized) == col(0: repo_name_normalized)| || * Pre-Shuffle Merge| | Threshold: 1073741824| || * Repartition: Random| | Num partitions = Some(5000)| || * Pre-Shuffle Merge| | Threshold: 1073741824| || * ScanTaskSource:| | Num Scan Tasks = 271| | Estimated Scan Bytes = 5317950853| | Pushdowns: {projection: [repo_name_normalized], filter:| | not(is_null(col(repo_name_normalized) == col(repo_name_normalized)))}| | Schema: {repo_name_normalized#Utf8, repo_name#Utf8}| | Scan Tasks: [| | {File {file:///mnt/test//v1/part-00000-cc7097a6-3100-4933-ac2c-| | 9050f71d73f4-c000.snappy.parquet}}| | {File {file:///mnt/test//v1/part-00001-cc7097a6-3100-4933-ac2c-| | 9050f71d73f4-c000.snappy.parquet}}| | {File {file:///mnt/test//v1/part-00002-cc7097a6-3100-4933-ac2c-| | 9050f71d73f4-c000.snappy.parquet}}| | ...| | {File {file:///mnt/test//v2/part-00236-5fb09fd6-1cbf-471f-9544-| | b92a0f18d8b0-c000.snappy.parquet}}| | {File {file:///mnt/test//v2/part-00237-5fb09fd6-1cbf-471f-9544-| | b92a0f18d8b0-c000.snappy.parquet}}| | {File {file:///mnt/test//v2/part-00238-5fb09fd6-1cbf-471f-9544-| | b92a0f18d8b0-c000.snappy.parquet}}| | ]|* Repartition: Hash| Num partitions = Some(5000)| By = col(2: repo_name_normalized) == col(2: repo_name_normalized)|* Pre-Shuffle Merge| Threshold: 1073741824|* Project: col(0: repo_name), col(1: stars), lower(replace(col(0: repo_name),| lit("[./\-]"), lit("_"))) as repo_name_normalized| Resource request = None|* Filter: col(1: stars) > lit(0)|* Project: get(col(0: content)) as repo_name, get(col(0: content)) as stars| Resource request = None|* Repartition: Random| Num partitions = Some(5000)|* Pre-Shuffle Merge| Threshold: 1073741824|* ScanTaskSource:| Num Scan Tasks = 119504| Estimated Scan Bytes = 418478051574| Schema: {content#Struct([Field { name: "allow_forking", dtype: Boolean,| metadata: {} }, Field { name: "archive_url", dtype: Utf8, metadata: {} },| Field { name: "archived", dtype: Boolean, metadata: {} }, Field { name:| "assignees_url", dtype: Utf8, metadata: {} }, Field { name: "blobs_url", dtype:| Utf8, metadata: {} }, Field { name: "branches_url", dtype: Utf8, metadata:| {} }, Field { name: "clone_url", dtype: Utf8, metadata: {} }, Field { name:| "collaborators_url", dtype: Utf8, metadata: {} }, Field { name: "comments_url",| dtype: Utf8, metadata: {} }, Field { name: "commits_url", dtype: Utf8, metadata:| {} }, Field { name: "compare_url", dtype: Utf8, metadata: {} }, Field { name:| "contents_url", dtype: Utf8, metadata: {} }, Field { name: "contributors_url",| dtype: Utf8, metadata: {} }, Field { name: "created_at", dtype: Utf8,| metadata: {} }, Field { name: "default_branch", dtype: Utf8, metadata: {} },| Field { name: "deployments_url", dtype: Utf8, metadata: {} }, Field { name:| "description", dtype: Utf8, metadata: {} }, Field { name: "disabled", dtype:| Boolean, metadata: {} }, Field { name: "downloads_url", dtype: Utf8, metadata:| {} }, Field { name: "events_url", dtype: Utf8, metadata: {} }, Field { name:| "fork", dtype: Boolean, metadata: {} }, Field { name: "forks", dtype: Int64,| metadata: {} }, Field { name: "forks_count", dtype: Int64, metadata: {} }, Field| { name: "forks_url", dtype: Utf8, metadata: {} }, Field { name: "full_name",| dtype: Utf8, metadata: {} }, Field { name: "git_commits_url", dtype: Utf8,| metadata: {} }, Field { name: "git_refs_url", dtype: Utf8, metadata: {} },| Field { name: "git_tags_url", dtype: Utf8, metadata: {} }, Field { name:| "git_url", dtype: Utf8, metadata: {} }, Field { name: "has_discussions",| dtype: Boolean, metadata: {} }, Field { name: "has_downloads", dtype: Boolean,| metadata: {} }, Field { name: "has_issues", dtype: Boolean, metadata: {} },| Field { name: "has_pages", dtype: Boolean, metadata: {} }, Field { name:| "has_projects", dtype: Boolean, metadata: {} }, Field { name: "has_wiki",| dtype: Boolean, metadata: {} }, Field { name: "homepage", dtype: Utf8, metadata:| {} }, Field { name: "hooks_url", dtype: Utf8, metadata: {} }, Field { name:| "html_url", dtype: Utf8, metadata: {} }, Field { name: "id", dtype: Int64,| metadata: {} }, Field { name: "is_template", dtype: Boolean, metadata: {} },| Field { name: "issue_comment_url", dtype: Utf8, metadata: {} }, Field { name:| "issue_events_url", dtype: Utf8, metadata: {} }, Field { name: "issues_url",| dtype: Utf8, metadata: {} }, Field { name: "keys_url", dtype: Utf8, metadata:| {} }, Field { name: "labels_url", dtype: Utf8, metadata: {} }, Field { name:| "language", dtype: Utf8, metadata: {} }, Field { name: "languages_url",| dtype: Utf8, metadata: {} }, Field { name: "license", dtype: Struct([Field| { name: "key", dtype: Utf8, metadata: {} }, Field { name: "name", dtype: Utf8,| metadata: {} }, Field { name: "node_id", dtype: Utf8, metadata: {} }, Field| { name: "spdx_id", dtype: Utf8, metadata: {} }, Field { name: "url", dtype:| Utf8, metadata: {} }]), metadata: {} }, Field { name: "merges_url", dtype:| Utf8, metadata: {} }, Field { name: "milestones_url", dtype: Utf8, metadata:| {} }, Field { name: "mirror_url", dtype: Utf8, metadata: {} }, Field { name:| "name", dtype: Utf8, metadata: {} }, Field { name: "node_id", dtype: Utf8,| metadata: {} }, Field { name: "notifications_url", dtype: Utf8, metadata:| {} }, Field { name: "open_issues", dtype: Int64, metadata: {} }, Field { name:| "open_issues_count", dtype: Int64, metadata: {} }, Field { name: "owner", dtype:| Struct([Field { name: "avatar_url", dtype: Utf8, metadata: {} }, Field { name:| "events_url", dtype: Utf8, metadata: {} }, Field { name: "followers_url", dtype:| Utf8, metadata: {} }, Field { name: "following_url", dtype: Utf8, metadata:| {} }, Field { name: "gists_url", dtype: Utf8, metadata: {} }, Field { name:| "gravatar_id", dtype: Utf8, metadata: {} }, Field { name: "html_url", dtype:| Utf8, metadata: {} }, Field { name: "id", dtype: Int64, metadata: {} }, Field| { name: "login", dtype: Utf8, metadata: {} }, Field { name: "node_id", dtype:| Utf8, metadata: {} }, Field { name: "organizations_url", dtype: Utf8, metadata:| {} }, Field { name: "received_events_url", dtype: Utf8, metadata: {} }, Field| { name: "repos_url", dtype: Utf8, metadata: {} }, Field { name: "site_admin",| dtype: Boolean, metadata: {} }, Field { name: "starred_url", dtype: Utf8,| metadata: {} }, Field { name: "subscriptions_url", dtype: Utf8, metadata:| {} }, Field { name: "type", dtype: Utf8, metadata: {} }, Field { name: "url",| dtype: Utf8, metadata: {} }, Field { name: "user_view_type", dtype: Utf8,| metadata: {} }]), metadata: {} }, Field { name: "private", dtype: Boolean,| metadata: {} }, Field { name: "pulls_url", dtype: Utf8, metadata: {} }, Field| { name: "pushed_at", dtype: Utf8, metadata: {} }, Field { name: "releases_url",| dtype: Utf8, metadata: {} }, Field { name: "score", dtype: Float64, metadata:| {} }, Field { name: "size", dtype: Int64, metadata: {} }, Field { name:| "ssh_url", dtype: Utf8, metadata: {} }, Field { name: "stargazers_count",| dtype: Int64, metadata: {} }, Field { name: "stargazers_url", dtype: Utf8,| metadata: {} }, Field { name: "statuses_url", dtype: Utf8, metadata: {} },| Field { name: "subscribers_url", dtype: Utf8, metadata: {} }, Field { name:| "subscription_url", dtype: Utf8, metadata: {} }, Field { name: "svn_url", dtype:| Utf8, metadata: {} }, Field { name: "tags_url", dtype: Utf8, metadata: {} },| Field { name: "teams_url", dtype: Utf8, metadata: {} }, Field { name: "topics",| dtype: List(Utf8), metadata: {} }, Field { name: "trees_url", dtype: Utf8,| metadata: {} }, Field { name: "updated_at", dtype: Utf8, metadata: {} }, Field| { name: "url", dtype: Utf8, metadata: {} }, Field { name: "visibility", dtype:| Utf8, metadata: {} }, Field { name: "watchers", dtype: Int64, metadata: {} },| Field { name: "watchers_count", dtype: Int64, metadata: {} }, Field { name:| "web_commit_signoff_required", dtype: Boolean, metadata: {} }])}| Scan Tasks: [| {File {file:///mnt/test_repo//20251014/part-| 00000-1f37045a-f911-4661-925c-7c145b6e46eb-c000.snappy.parquet}}| {File {file:///mnt/test_repo//20251014/part-| 00001-1f37045a-f911-4661-925c-7c145b6e46eb-c000.snappy.parquet}}| {File {file:///mnt/test_repo//20251014/part-| 00002-1f37045a-f911-4661-925c-7c145b6e46eb-c000.snappy.parquet}}| ...| {File {file:///mnt/test_repo//20250930/part-| 08033-746def56-b636-4915-aba8-684e38c89cfe-c000.snappy.parquet}}| {File {file:///mnt/test_repo//20250930/part-| 08034-746def56-b636-4915-aba8-684e38c89cfe-c000.snappy.parquet}}| {File {file:///mnt/test_repo//20250930/part-| 08035-746def56-b636-4915-aba8-684e38c89cfe-c000.snappy.parquet}}| ]c
Colin Ho
11/18/2025, 12:54 PMwhen you do the repartition can you also provide the join key for the partition_by? i.e.
repartition(5000, "repo_name_normalized")Colin Ho
11/18/2025, 12:56 PMwithout the column, it will do a random repartition. Instead we want to do hash based repartition on the join_key, then this will be used for the subsequent hash join
c
can cai
11/18/2025, 1:04 PMgot it，let me try it.
can cai
11/18/2025, 1:08 PMHi @Colin Ho i submitted a job, but the running parallelism is very low.
There are only one or two running tasks at the same time, and is there any restriction on concurrency somewhere?
c
Colin Ho
11/18/2025, 1:14 PMoh
Colin Ho
11/18/2025, 1:15 PMray dashboard can only show max 10000 tasks
Colin Ho
11/18/2025, 1:15 PMonce past that it is completely innacurate
Colin Ho
11/18/2025, 1:15 PMColin Ho
11/18/2025, 1:16 PMColin Ho
11/18/2025, 1:16 PMit's a known issue for them
c
can cai
11/18/2025, 1:20 PMAs far as I know, if there is running state tasks, ray gcs seems that it will evict the old ones (such as finished, failed) and display them again on the dashboard.
So if the number of running tasks is really large, it shouldn't just display a few. WDYT? cc @Colin Ho
can cai
11/18/2025, 1:23 PMAnother issue encountered before, is that when running on flotilla, shuffle takes up too much memory and is prone to causing oom killed. Now it has occurred again, as follows:
Task was killed due to the node running low on memory.PID MEM(GB) COMMAND263 17.19 ray::PhysicalScan->Project->Filter->Projectc
Colin Ho
11/18/2025, 1:58 PMNo i don't think so, take this example:
Copy code
import ray
import time
@ray.remote
def sleep(x):
time.sleep(x)
ray.get([sleep.remote(0.1) for _ in range(20000)])Colin Ho
11/18/2025, 1:58 PMc
can cai
11/18/2025, 2:45 PMThe task event capacity in gcs is 100k, and it will only be evicted when it reaches 100k.
And the default display capacity of the dashboard ui is 10k. like this:
https://github.com/ray-project/ray/blob/61905b8ef89c43aeb03abbb62e81c3030773bb53/src/ray/gcs/gcs_task_manager.h#L170-L174
https://github.com/ray-project/ray/blob/61905b8ef89c43aeb03abbb62e81c3030773bb53/src/ray/common/ray_config_def.h#L475-L478
However，maybe i can increase the value of
RAY_MAX_LIMIT_FROM_DATA_SOURCERAY_MAX_LIMIT_FROM_API_SERVERcan cai
11/18/2025, 2:50 PMBut the problem that troubles me the most at present is both the driver and workers consume a large amount of memory, which may lead to oom, especially for shuffle.
c
Colin Ho
11/18/2025, 11:18 PMUnfortunately your shuffle and join are quite large, and I don't think there's anything else we can do immediately here. We just need to build a better shuffle that can handle this.
Colin Ho
11/18/2025, 11:19 PMSorry about this
c
can cai
11/19/2025, 2:29 AMThank you for your patient reply.