Failed to run Document Embedding Benchmarking code

[wpjgit]

Describe the bug

failed to load region from IMDS err=failed to load IMDS session token: dispatch failure: timeout: client error (Connect): HTTP connect timeout occurred after 1s: timed out (FailedToLoadToken(FailedToLoadToken { source: DispatchFailure(DispatchFailure { source: ConnectorError { kind: Timeout, source: hyper_util::client::legacy::Error(Connect, HttpTimeoutError { kind: "HTTP connect", duration: 1s }), connection: Unknown } }) }))
S3 Credentials could not be loaded. Reverting to Anonymous mode. the credential provider was not enabled

When I ran this code on my Windows computer, I encountered this issue. Is there any way to resolve it successfully?

To Reproduce

No response

Expected behavior

No response

Component(s)

Parquet

Additional context

No response

[colin-ho]

Try setting this?

import daft

s3_config = daft.io.S3Config.from_env()
s3_config = s3_config.replace(requester_pays=True)
daft.set_planning_config(daft.io.IOConfig(s3=s3_config))

[wpjgit]

Try setting this?

import daft

s3_config = daft.io.S3Config.from_env()
s3_config = s3_config.replace(requester_pays=True)
daft.set_planning_config(daft.io.IOConfig(s3=s3_config))

thanks，I tried your method, but the same error still occurred.

[wpjgit]

The s3 File Access Denied

[colin-ho]

The bucket is a requester pays bucket, one more thing you can try is just manually pass in the credentials

s3_config = S3Config(
    requester_pays=True,
    anonymous=False,
    region_name="us-west-2",
    session_token=AWS_SESSION_TOKEN,
    key_id=AWS_ACCESS_KEY_ID,
    access_key=AWS_SECRET_ACCESS_KEY
)
IO_CONFIG = IOConfig(s3=s3_config)
daft.set_planning_config(default_io_config=IO_CONFIG)

Also do make sure that you have the IAM credentials set to access a requester pays bucket

[brucewkz]

@colin-ho

this is my code:

    s3 = boto3.client('s3', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name=region)        
    paginator = s3.get_paginator('list_objects_v2')
    pages = paginator.paginate(Bucket=bucket_name, Prefix=s3_prefix)

    pdf_files = []
    for page in pages:
        if 'Contents' in page:
            for obj in page['Contents']:                   
                if obj['Key'].lower().endswith('.pdf'):
                    pdf_files.append(obj['Key'])

throw the following errorr:
An error occurred (AccessDenied) when calling the ListObjectsV2 operation: Access Denied

Can you show your code?

[colin-ho]

pages = paginator.paginate(
    Bucket=bucket_name,
    Prefix=s3_prefix,
    RequestPayer='requester'
)

Make sure to set request payer as requester, because the bucket is a "requester-pays" bucket https://docs.aws.amazon.com/AmazonS3/latest/userguide/RequesterPaysBuckets.html

This is the code I recommend:

s3_config = S3Config(
    requester_pays=True, # this is the important part
    anonymous=False,
    region_name="us-west-2",
    session_token=AWS_SESSION_TOKEN,
    key_id=AWS_ACCESS_KEY_ID,
    access_key=AWS_SECRET_ACCESS_KEY
)
IO_CONFIG = IOConfig(s3=s3_config)
daft.set_planning_config(default_io_config=IO_CONFIG)

[wpjgit]

I have prepared the data.

But，I can run it successfully using ray_data_main.py, but it fails when using daft_main.py, and the error is: Segmentation fault (core dumped)

My Cluster configuration
2 instances, each with:
8 NVIDIA A100 GPU, 32 GB GPU memory
104 vCPUs, 786 GB memory
1 TB volume

And，after I added the parameter use_process=True to extract_text_from_parsed_pdf, daft_main could run successfully(do not use ray for execution)

[wpjgit]

I have prepared the data.

But，I can run it successfully using ray_data_main.py, but it fails when using daft_main.py, and the error is: Segmentation fault (core dumped)

My Cluster configuration 2 instances, each with: 8 NVIDIA A100 GPU, 32 GB GPU memory 104 vCPUs, 786 GB memory 1 TB volume

And，after I added the parameter use_process=True to extract_text_from_parsed_pdf, daft_main could run successfully(do not use ray for execution)

execute daft_main.py error log:
ERROR1 ->:
(RaySwordfishActor pid=1187021, ip=10.44.20.99) free(): invalid next size (fast)
(RaySwordfishActor pid=1187021, ip=10.44.20.99) *** SIGABRT received at time=1760947744 on cpu 86 ***
(RaySwordfishActor pid=1187021, ip=10.44.20.99) PC: @ 0x7fad27c5700b (unknown) raise
(RaySwordfishActor pid=1187021, ip=10.44.20.99) @ 0x7fad27f74420 4176 (unknown)
(RaySwordfishActor pid=1187021, ip=10.44.20.99) [2025-10-20 16:09:04,718 E 1187021 1199730] logging.cc:474: *** SIGABRT received at time=1760947744 on cpu 86 ***
(RaySwordfishActor pid=1187021, ip=10.44.20.99) [2025-10-20 16:09:04,718 E 1187021 1199730] logging.cc:474: PC: @ 0x7fad27c5700b (unknown) raise
(RaySwordfishActor pid=1187021, ip=10.44.20.99) [2025-10-20 16:09:04,721 E 1187021 1199730] logging.cc:474: @ 0x7fad27f74420 4176 (unknown)
(RaySwordfishActor pid=1187021, ip=10.44.20.99) Fatal Python error: Aborted
(RaySwordfishActor pid=1187021, ip=10.44.20.99)
(RaySwordfishActor pid=1187021, ip=10.44.20.99)
(RaySwordfishActor pid=1187021, ip=10.44.20.99) @ 0x1 (unknown) (unknown) [repeated 2x across cluster]
(RaySwordfishActor pid=1187021, ip=10.44.20.99) [2025-10-20 16:09:04,724 E 1187021 1199730] logging.cc:474: @ 0x1 (unknown) (unknown) [repeated 2x across cluster]
(RaySwordfishActor pid=1187021, ip=10.44.20.99) Stack (most recent call first):
(RaySwordfishActor pid=1187021, ip=10.44.20.99) File "/opt/anaconda3/envs/daft_benchmark/lib/python3.12/site-packages/daft/expressions/expressions.py", line 1234 in batch_func
(RaySwordfishActor pid=1187021, ip=10.44.20.99) File "/opt/miniconda3/envs/daft_benchmark/lib/python3.12/site-packages/daft/udf/legacy.py", line 165 in run_udf
(RaySwordfishActor pid=1187021, ip=10.44.20.99) Extension modules: msgpack._cmsgpack, google._upb._message, psutil._psutil_linux, psutil._psutil_posix, yaml._yaml, zstandard.backend_c, charset_normalizer.md, requests.packages.charset_normalizer.md, requests.packages.chardet.md, uvloop.loop, ray._raylet, numpy.core._multiarray_umath, numpy.core._multiarray_tests, numpy.linalg._umath_linalg, numpy.fft._pocketfft_internal, numpy.random._common, numpy.random.bit_generator, numpy.random._bounded_integers, numpy.random._mt19937, numpy.random.mtrand, numpy.random._philox, numpy.random._pcg64, numpy.random._sfc64, numpy.random._generator, pyarrow.lib, pyarrow._json, pymupdf._extra, pymupdf._mupdf (total: 28)

ERROR2 ->:
(RaySwordfishActor pid=1074113) *** SIGSEGV received at time=1760948283 on cpu 54 ***
(RaySwordfishActor pid=1074113) PC: @ 0x7f4fe6106f60 (unknown) fz_do_catch
(RaySwordfishActor pid=1074113) [2025-10-20 16:18:03,824 E 1074113 1092343] logging.cc:474: *** SIGSEGV received at time=1760948283 on cpu 54 ***
(RaySwordfishActor pid=1074113) [2025-10-20 16:18:03,824 E 1074113 1092343] logging.cc:474: PC: @ 0x7f4fe6106f60 (unknown) fz_do_catch
(RaySwordfishActor pid=1074113) Fatal Python error: Segmentation fault
(RaySwordfishActor pid=1074113)
(RaySwordfishActor pid=1074113) Stack (most recent call first):
(RaySwordfishActor pid=1074113) File "/opt/anaconda3/envs/daft_benchmark/lib/python3.12/site-packages/pymupdf/extra.py", line 195 in page_get_textpage
(RaySwordfishActor pid=1074113) File "/opt/anaconda3/envs/daft_benchmark/lib/python3.12/site-packages/pymupdf/init.py", line 8043 in _get_textpage
(RaySwordfishActor pid=1074113) File "/opt/anaconda3/envs/daft_benchmark/lib/python3.12/site-packages/pymupdf/init.py", line 9514 in get_textpage
(RaySwordfishActor pid=1074113) File "/opt/anaconda3/envs/daft_benchmark/lib/python3.12/site-packages/pymupdf/utils.py", line 983 in get_text
(RaySwordfishActor pid=1074113) File "/opt/software/daft-benchmark/ai/document_embedding/daft_main.py", line 42 in extract_text_from_parsed_pdf
(RaySwordfishActor pid=1074113) File "/opt/anaconda3/envs/daft_benchmark/lib/python3.12/site-packages/daft/expressions/expressions.py", line 1234 in batch_func
(RaySwordfishActor pid=1074113) File "/opt/anaconda3/envs/daft_benchmark/lib/python3.12/site-packages/daft/udf/legacy.py", line 165 in run_udf

[malcolmgreaves]

Hi @wpjgit do you have any code we can look at?

The SIGSEGV and stack trace from pymupdf shows that this specific problem is coming from outside of Daft (https://github.com/pymupdf/PyMuPDF/blob/main/src/utils.py). It may be that there is a bug in combination of PDF you're trying to parse and version of pymupdf. Or, you may be running into a race condition with the way you're using PyMuPDF.

[wpjgit]

My code is from https://github.com/Eventual-Inc/Daft/blob/main/benchmarking/ai/document_embedding/daft_main.py

The input data consists of 10,000 PDFs that I downloaded from Common Crawl.

I don't understand why ray data can succeed but daft fails, and when I set use_process=True for the UDF, daft works successfully. I wonder if there might be some issues with daft's parallel processing?

udf code:

...

CHUNK_OVERLAP = 200
EMBEDDING_BATCH_SIZE = 10

# Not running on Ray
# daft.context.set_runner_ray()

# Wait for Ray cluster to be ready
@ray.remote
def warmup():
    pass
# ray.get([warmup.remote() for _ in range(64)])


@daft.func(return_dtype=daft.DataType.list(daft.DataType.struct({"text": daft.DataType.string(), "page_number": daft.DataType.int64()})), use_process=True)
def extract_text_from_parsed_pdf(pdf_bytes):
    try:
        doc = pymupdf.Document(stream=pdf_bytes, filetype="pdf")
        if len(doc) > MAX_PDF_PAGES:
            print(f"Skipping PDF because it has {len(doc)} pages")
            return None
        page_texts = [{"text": page.get_text(), "page_number": page.number} for page in doc]
        return page_texts
    except Exception as e:
        print(f"Error extracting text from PDF {e}")
        return None
...

df = daft.read_parquet(INPUT_PATH)
df = df.where(daft.col("file_name").str.endswith(".pdf"))
df = df.with_column("pdf_bytes", df["uploaded_pdf_path"].url.download())
df = df.with_column("pages", extract_text_from_parsed_pdf(df["pdf_bytes"]))  # change the calling method
df = df.explode("pages")
df = df.with_columns({"page_text": col("pages")["text"], "page_number": col("pages")["page_number"]})

...