Setting AWS input files

[riazarbi]

Hello,

TL;DR - Is it currently possible to set a tar_target to read input data from an unsupported file sitting in AWS (parquet in my instance)?

I'm trying to set input file target that sits on aws s3. By way of example, if you take the canonical minimal example,

# _targets.R
library(targets)
path_to_data <- function() {
  "data/raw_data.csv"
}
list(
  tar_target(
    raw_data_file,
    path_to_data(),
    format = "file"
  ),
  tar_target(
    raw_data,
    read_csv(raw_data_file, col_types = cols())
  )
)

I would like to modify the above code so that I am referencing an unsuported file format in AWS. Like so -

# _targets.R
library(targets)
path_to_data <- function() {
  "raw_data.parquet"
}
list(
  tar_target(
    raw_data_file,
    path_to_data(),
    format = "aws_file",
   resources = list(bucket ="raw_data_bucket")
  ),
  tar_target(
    raw_data,
    read_parquet(raw_data_file)
  )
)

Now, I know that my aws.s3 creds are working because I can use aws_fst as a _targets storage backend. But when I run the above code (or similar) I get -

● run target raw_data
x error target raw_data
Error : IOError: Failed to open local file 'raw_data.parquet'. Detail: [errno 2] No such file or directory
Error: callr subprocess failed: IOError: Failed to open local file 'data.parquet'. Detail: [errno 2] No such file or directory`

I can get the data by just using aws.s3::save_object in a tar_target but I lose out on all that "watch for upstream changes" goodness.

So. Am I trying to do something that is not actually possible?

[riazarbi]

So I can see that if I create a file in my local environment called data.parquet, tar_make() on the above _targets.R file will take that file, upload it to raw_data_bucket/_targets/objects/raw_data_file and then monitor it for downstream activity.

But my use case is different - in my situation there is already a parquet file in a bucket, which is refreshed periodically, that I want to read into my targets folder.

[riazarbi]

Ok, @wlandau I went down a rabbit hole, poked around your source, and solved my problem.

If I comment out line 50 of utils_url.R, then my tar_make() runs successfully.

url_handle <- function(handle = NULL) {
  handle <- handle %||% curl::new_handle()
  assert_inherits(handle, "curl_handle")
  #curl::handle_setopt(handle, nobody = TRUE)
  handle
}

It seems to me (and I could totally be wrong) that curl::handle_setopt will just replace those options named, while curl::handle_setheaders replaces the entire header set.

So the headers I am passing in are not replacing the header set completely.

I looked at trying to change this line of code curl::handle_setopt(handle, nobody = TRUE), but then I noticed that you are actually using the handle=handle option pretty much everywhere in the script.

So, my question to you - is that line actually needed? Maybe it can be deleted?

[riazarbi]

The way I see it, removing that line ensures that the handle sent in via the resources parameter will be all and only what gets passed to the curl call, which is desirable.

One must just be sure that all curl calls in the package are accompanied by a handle parameter and preceded by a url_handle call.

[wlandau]

We only want the headers, and nobody = TRUE ensures we do not accidentally download the whole body of the file ("no body"). I did not realize this would cause problems.

[wlandau]

I agree with your suggestion because it prevents surprises like the one you described. As of ae0f8f6, the default handle has nobody = TRUE and custom handles are not modified. The docs now urge people to set nobody = TRUE in their own custom handles.

[riazarbi]

Bellissimo!

[riazarbi]

Here, for posterity, is my setup. This allows me to use the "url" format on minio-hosted S3 objects. The url path construction may need to be modified for it to work with S3 subdomain-style bucket paths, YMMV.

Will only work after ae0f8f6.

It assumes that environment variables AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY are set.

In R/functions.R

create_s3_handle <- function(object_url) {
  domain <-
    function(x)
      strsplit(gsub("http://|https://|www\\.", "", x), "/")[[c(1, 1)]]
  bucket <-
    function(x)
      strsplit(gsub("http://|https://|www\\.", "", x), "/")[[c(1, 2)]]
  object_name <- function(x) {
    y <- gsub("http://|https://|www\\.", "", x)
    z <-
      stringr::str_remove(y, paste0(domain(x), "/"))
      stringr::str_remove(z, paste0(bucket(x), "/"))
    
  }
  
  date_time <- format(Sys.time(), "%Y%m%dT%H%M%SZ", tz = "UTC")
  curl::handle_setheaders(
    curl::new_handle(),
    .list = list(
      Host = domain(object_url),
      Date = date_time,
      `Content-Type` = "application/octet-stream",
      Authorization = paste0(
        "AWS ",
        Sys.getenv("AWS_ACCESS_KEY_ID"),
        ":",
        RCurl::base64(digest::hmac(
          Sys.getenv("AWS_SECRET_ACCESS_KEY"),
          paste0(
            "GET\n\n",
            "application/octet-stream",
            "\n",
            date_time,
            "\n/",
            bucket(object_url),
            "/",
            object_name(object_url)
          ),
          algo = "sha1",
          raw = TRUE
        ))
      )
    )
  )
}

My _targets.R file:

library(targets)

source("R/functions.R")

tar_option_set(
  packages = c(
    "arrow",
    "dplyr",
    "stringr",
    "curl",
    "digest",
    "RCurl"),
  format = "fst_tbl",
  memory = "persistent",
  error = "stop"
)


# set input paths
path_to_aws_object <- "https://MINIO_HOST/MINIO_BUCKET/tmp.parquet"

list(
  # example of a live make-like input conneciton
     tar_target(
       aws_url,
       path_to_aws_object,
       format = "url",
       resources = list(handle = create_s3_handle(path_to_aws_object))
     ),
   tar_target(
     parquet_df, {
       tf <- tempfile()
       curl_download(aws_url, 
                     tf, 
                     handle = create_s3_handle(aws_url)
                     )
       read_parquet(tf)
       unlink(tf)
       
     }
   )
)