Exporta archivos a Cloudflare R2 con Scala y el SDK de AWS
Usa AWS SDK for Java v2 para exportar archivos locales terminados desde una aplicación Scala a un bucket privado de Cloudflare R2. Este ejemplo ofrece un programa de línea de comandos con modos de subida síncrona, asíncrona, en streaming con FS2 y en paralelo, además de una forma de verificar los bytes almacenados.
¿Por qué elegir Cloudflare R2?
La API compatible con S3 de R2 permite que el código Scala use la firma de solicitudes y los clientes HTTP del SDK de Java. El ejemplo utiliza un bucket sin restricción de jurisdicción en el endpoint estándar de la cuenta. Los buckets de una jurisdicción específica necesitan el endpoint correspondiente.
Configura tu proyecto Scala
Esta guía está dirigida a aplicaciones existentes de Scala 2.13 que usan sbt 1.x. Las versiones fijadas y probadas son Scala 2.13.18, sbt 1.13.0, Cats Effect 3.7.1, FS2 3.14.0 y AWS SDK 2.56.0. Scala 3 y sbt 2 son opciones de migración independientes; estas versiones mantienen el ejemplo dentro de la familia de compilación de Scala 2.13.
Usa Bash, cURL y una instalación de OpenJDK 21 con mantenimiento,
con java en tu PATH. El parche del JDK probado
es 21.0.12.1; usa la actualización de seguridad vigente de tu distribuidor del JDK dentro de la
versión 21. No necesitas instalar por separado un compilador de Scala ni el comando sbt. Ejecuta
lo siguiente desde un directorio con permisos de escritura para crear un proyecto nuevo; la
subshell deja tu directorio actual sin cambios:
(
set -eu
command -v java >/dev/null
command -v curl >/dev/null
java -version
mkdir r2-scala
cd r2-scala
mkdir -p project src/main/scala
curl -fsSLo sbt-launch.jar https://repo.maven.apache.org/maven2/org/scala-sbt/sbt-launch/1.13.0/sbt-launch-1.13.0.jar
)
Si la creación falla porque r2-scala ya existe, elige otro directorio padre.
Si la descarga falla, conserva el proyecto y vuelve a ejecutar solo el comando cURL desde su
interior. Guarda los siguientes archivos en r2-scala. Usa
project/build.properties para fijar la versión del lanzador:
sbt.version=1.13.0
Guarda build.sbt:
scalaVersion := "2.13.18"
libraryDependencies ++= Seq(
"software.amazon.awssdk" % "s3" % "2.56.0",
"software.amazon.awssdk" % "netty-nio-client" % "2.56.0",
"software.amazon.awssdk" % "apache-client" % "2.56.0",
"org.typelevel" %% "cats-effect" % "3.7.1",
"co.fs2" %% "fs2-io" % "3.14.0",
"co.fs2" %% "fs2-reactive-streams" % "3.14.0"
)
Compile / run / fork := true
Configura el cliente del SDK de AWS
Crea un bucket y credenciales S3 de R2 cuyo ámbito se limite a
ese bucket, con permisos de lectura y escritura de objetos. Proporciona
R2_ACCOUNT_ID, R2_ACCESS_KEY_ID,
R2_SECRET_ACCESS_KEY y R2_BUCKET mediante las variables de entorno
del proceso antes de ejecutar el programa. Usa un gestor de secretos o la configuración del entorno
de tu terminal; no incluyas las credenciales en el archivo Scala ni en los argumentos del comando.
El bucket permanece privado y debe existir de antemano.
Los clientes usan la región auto, direccionamiento de estilo de ruta y
codificación fragmentada desactivada, siguiendo la
configuración de Java de Cloudflare.
WHEN_REQUIRED desactiva las
sumas de comprobación opcionales de las solicitudes del SDK.
El programa selecciona Apache para HTTP síncrono y Netty para HTTP asíncrono, y cierra cada cliente
con Resource de Cats Effect.
Subida síncrona básica
Elige sync para realizar un PUT bloqueante sencillo. Lee el archivo
mediante RequestBody.fromFile en el pool de hilos para operaciones bloqueantes de Cats
Effect. La cancelación espera a que la llamada bloqueante termine o agote su tiempo de espera;
elige async o fs2 cuando necesites cancelar
el futuro subyacente.
Subida asíncrona con Cats Effect
Guarda este programa completo como src/main/scala/R2Upload.scala.
async usa el cuerpo basado en archivo del SDK;
fs2 proporciona un publicador de Reactive Streams. Ambos esperan
mediante el puente de CompletableFuture de Cats Effect, que
solicita la cancelación del futuro cuando se cancela su fibra.
import cats.effect.{ExitCode, IO, IOApp, Resource}
import cats.effect.implicits._
import cats.syntax.all._
import fs2.interop.reactivestreams._
import fs2.io.file.{Files, Flags, Path}
import java.io.InputStream
import java.net.URI
import java.nio.ByteBuffer
import java.security.MessageDigest
import java.time.Duration
import java.util.Optional
import org.reactivestreams.Subscriber
import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
import software.amazon.awssdk.core.async.AsyncRequestBody
import software.amazon.awssdk.core.checksums.RequestChecksumCalculation
import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration
import software.amazon.awssdk.core.sync.RequestBody
import software.amazon.awssdk.http.apache.ApacheHttpClient
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.retries.StandardRetryStrategy
import software.amazon.awssdk.services.s3.{S3AsyncClient, S3Client, S3Configuration}
import software.amazon.awssdk.services.s3.model.{GetObjectRequest, PutObjectRequest, PutObjectResponse, S3Exception}
object R2Upload extends IOApp {
final case class Config(endpoint: URI, accessKey: String, secretKey: String, bucket: String)
final case class Task(path: Path, key: String)
private val service = S3Configuration.builder()
.pathStyleAccessEnabled(true)
.chunkedEncodingEnabled(false)
.build()
private def overrides: ClientOverrideConfiguration =
ClientOverrideConfiguration.builder()
.apiCallTimeout(Duration.ofMinutes(2))
.apiCallAttemptTimeout(Duration.ofSeconds(45))
.retryStrategy(StandardRetryStrategy.builder().maxAttempts(3).build())
.build()
private def credentials(config: Config): StaticCredentialsProvider =
StaticCredentialsProvider.create(
AwsBasicCredentials.create(config.accessKey, config.secretKey)
)
def asyncClient(config: Config): Resource[IO, S3AsyncClient] =
Resource.fromAutoCloseable(IO.blocking {
S3AsyncClient.builder()
.httpClientBuilder(NettyNioAsyncHttpClient.builder()
.connectionTimeout(Duration.ofSeconds(10))
.readTimeout(Duration.ofSeconds(30))
.writeTimeout(Duration.ofSeconds(30)))
.endpointOverride(config.endpoint)
.credentialsProvider(credentials(config))
.region(Region.of("auto"))
.serviceConfiguration(service)
.requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
.overrideConfiguration(overrides)
.build()
})
def syncClient(config: Config): Resource[IO, S3Client] =
Resource.fromAutoCloseable(IO.blocking {
S3Client.builder()
.httpClientBuilder(ApacheHttpClient.builder()
.connectionTimeout(Duration.ofSeconds(10))
.socketTimeout(Duration.ofSeconds(30)))
.endpointOverride(config.endpoint)
.credentialsProvider(credentials(config))
.region(Region.of("auto"))
.serviceConfiguration(service)
.requestChecksumCalculation(RequestChecksumCalculation.WHEN_REQUIRED)
.overrideConfiguration(overrides)
.build()
})
private def request(bucket: String, task: Task, size: Long): PutObjectRequest = {
require(size >= 0 && size <= 5L * 1024 * 1024 * 1024,
"Use multipart upload for files larger than 5 GiB")
PutObjectRequest.builder()
.bucket(bucket).key(task.key)
.contentType("application/octet-stream")
.contentLength(size)
.build()
}
def uploadSync(client: S3Client, bucket: String, task: Task): IO[PutObjectResponse] =
Files[IO].size(task.path).flatMap { size =>
IO.blocking(client.putObject(
request(bucket, task, size), RequestBody.fromFile(task.path.toNioPath)
))
}
def uploadAsync(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
Files[IO].size(task.path).flatMap { size =>
IO.fromCompletableFuture(IO {
client.putObject(request(bucket, task, size),
AsyncRequestBody.fromFile(task.path.toNioPath))
})
}
def uploadStream(client: S3AsyncClient, bucket: String, task: Task): IO[PutObjectResponse] =
Files[IO].size(task.path).flatMap { size =>
val buffers = Files[IO].readAll(task.path, 64 * 1024, Flags.Read)
.chunks.map(chunk => ByteBuffer.wrap(chunk.toArray))
buffers.toUnicastPublisher.use { publisher =>
val body = new AsyncRequestBody {
override def contentLength(): Optional[java.lang.Long] =
Optional.of(java.lang.Long.valueOf(size))
override def subscribe(subscriber: Subscriber[_ >: ByteBuffer]): Unit =
publisher.subscribe(subscriber)
}
IO.fromCompletableFuture(IO {
client.putObject(request(bucket, task, size), body)
})
}
}
def uploadMany(client: S3AsyncClient, bucket: String,
tasks: List[Task]): IO[List[PutObjectResponse]] =
tasks.parTraverseN(4)(task => uploadAsync(client, bucket, task))
private def digest(input: InputStream): Array[Byte] = {
val hash = MessageDigest.getInstance("SHA-256")
val buffer = new Array[Byte](64 * 1024)
var count = input.read(buffer)
while (count != -1) {
hash.update(buffer, 0, count)
count = input.read(buffer)
}
hash.digest()
}
def verify(client: S3Client, bucket: String, task: Task): IO[Unit] = {
val local = Resource.fromAutoCloseable(IO.blocking(
java.nio.file.Files.newInputStream(task.path.toNioPath)))
val remote = Resource.fromAutoCloseable(IO.blocking(client.getObject(
GetObjectRequest.builder().bucket(bucket).key(task.key).build())))
(local, remote).tupled.use { case (source, stored) =>
IO.blocking {
require(MessageDigest.isEqual(digest(source), digest(stored)), "Stored bytes differ")
}
}
}
override def run(args: List[String]): IO[ExitCode] = {
val operation = IO {
require(args.length >= 3 && args.length % 2 == 1,
"Usage: R2Upload sync|async|fs2|parallel|verify file key [file key ...]")
val mode = args.head
require(Set("sync", "async", "fs2", "parallel", "verify").contains(mode), "Unknown mode")
require(mode == "parallel" || args.length == 3, "Only parallel accepts multiple files")
val account = sys.env("R2_ACCOUNT_ID")
require(account.matches("[a-fA-F0-9]{32}"), "Invalid R2 account ID")
val config = Config(
URI.create(s"https://$account.r2.cloudflarestorage.com"),
sys.env("R2_ACCESS_KEY_ID"), sys.env("R2_SECRET_ACCESS_KEY"), sys.env("R2_BUCKET")
)
val tasks = args.tail.grouped(2).map(pair => Task(Path(pair.head), pair(1))).toList
require(tasks.forall(task =>
task.path.toString.matches("[A-Za-z0-9_./-]+") &&
task.key.matches("[A-Za-z0-9_./-]+") && !task.key.startsWith("-")),
"Use ASCII paths and keys containing only letters, digits, _, ., /, and -")
require(tasks.map(_.key).distinct.size == tasks.size, "Use distinct destination keys")
(mode, config, tasks)
}.flatMap { case (mode, config, tasks) =>
val upload = if (mode == "verify")
syncClient(config).use(client => verify(client, config.bucket, tasks.head))
else if (mode == "sync")
syncClient(config).use(client => uploadSync(client, config.bucket, tasks.head).void)
else
asyncClient(config).use { client =>
mode match {
case "fs2" => uploadStream(client, config.bucket, tasks.head).void
case "parallel" => uploadMany(client, config.bucket, tasks).void
case _ => uploadAsync(client, config.bucket, tasks.head).void
}
}
upload *> IO.println(if (mode == "verify") "Stored bytes match." else "Upload completed.")
}
operation.as(ExitCode.Success).handleErrorWith {
case error: S3Exception =>
IO.println(s"Upload failed (${error.getClass.getSimpleName}, HTTP ${error.statusCode()}).").as(ExitCode.Error)
case error =>
IO.println(s"Upload failed (${error.getClass.getSimpleName}). Check arguments, environment, file access, and network.")
.as(ExitCode.Error)
}
}
}
Abre una terminal en r2-scala. Estos comandos crean un archivo de entrada
pequeño, compilan el programa y lo suben. La primera compilación descarga las dependencias. Los
argumentos de archivo y clave de este lanzador se limitan a letras ASCII, dígitos,
_, .,
/ y -; usa
./ antes de un nombre local que empiece con un guion. Las claves de
destino no deben empezar con - y deben ser distintas dentro de un
mismo lote. Mantén los archivos de origen sin cambios hasta que terminen todos los intentos.
(
set -eu
printf 'Hello, R2!\n' > hello.txt
java -Dsbt.server.autostart=false -jar sbt-launch.jar compile
java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload async ./hello.txt uploads/hello.txt"
)
Reemplaza async por sync o
fs2 para usar los otros modos de archivo único. Un PUT exitoso reemplaza
cualquier objeto que tenga la misma clave, incluso cuando el archivo de entrada está vacío. Usa
una clave nueva si necesitas conservar el objeto anterior.
Después de una subida exitosa, vuelve a leer el objeto privado y compara los hashes SHA-256 del contenido completo del archivo local y del objeto almacenado:
java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload verify ./hello.txt uploads/hello.txt"
El verificador imprime Stored bytes match. solo cuando ambos hashes coinciden, y
devuelve un estado distinto de cero si no coinciden o si falla la lectura. Por sí solo, un PUT
completado confirma que el servicio lo aceptó; ejecuta la comparación antes de eliminar un original
importante. La verificación también requiere que el archivo de origen permanezca sin cambios.
Manejo de respuestas y errores
Upload completed. aparece solo después de que se completen todas las subidas
seleccionadas y termine el ámbito de uso del cliente. Los errores del servicio devuelven un código
de salida distinto de cero y un estado HTTP. Los errores de configuración local, archivos y red
también devuelven un código de salida distinto de cero, sin imprimir credenciales ni respuestas
sin procesar del SDK.
Una subida cancelada no cuenta como una finalización exitosa. Una cancelación o una respuesta perdida aún pueden dejar un objeto confirmado en el destino: inspecciónalo antes de volver a intentarlo. Este programa no revierte los objetos ni elimina datos remotos en caso de error.
Lógica de reintentos con espera exponencial
La estrategia estándar de reintentos del SDK maneja los errores que permiten reintentos, con un máximo de tres intentos por PUT. El tiempo de espera de la llamada a la API es de dos minutos, incluidos los reintentos; el de cada intento es de 45 segundos. El establecimiento de la conexión tiene un límite de 10 segundos. Estos límites se aplican a las solicitudes y no constituyen un plazo de dos minutos para un lote paralelo completo. Los límites de lectura de 30 segundos de los transportes son inferiores al tiempo de espera de cada intento; Netty también tiene un límite de escritura de 30 segundos.
Una transferencia lenta puede agotar el tiempo de espera incluso si no supera el límite de tamaño de archivo. Revisa el destino y luego ajusta los límites según tu conexión antes de volver a intentar subir el archivo sin cambios. No agregues un bucle de reintentos sin restricciones. El publicador de FS2 puede volver a ejecutar su flujo de archivo para una nueva suscripción, por lo que también necesita un archivo de origen estable que pueda volver a leerse.
Sube muchos archivos en paralelo
El modo parallel ejecuta como máximo cuatro efectos de subida a la vez.
Si uno falla, cancela los efectos hermanos y devuelve un estado de error. Los objetos completados
se conservan; un lote no es una transacción. Después de crear un segundo archivo local llamado
report.pdf, ejecuta:
java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload parallel ./hello.txt uploads/hello.txt ./report.pdf uploads/report.pdf"
El límite abarca las subidas activas, no la lista de argumentos en memoria ni todos los búferes internos de Netty y del SDK. No divide un archivo en partes paralelas.
Transmite archivos grandes con fs2
fs2 lee bloques de 64 KiB y los expone como elementos de tipo
ByteBuffer según la demanda del consumidor. El recurso del publicador
permanece abierto hasta que la operación se complete, falle o se cancele. Esto evita cargar el
archivo completo en un array de bytes; no impone un límite fijo a la memoria total de la JVM ni
del cliente HTTP.
Todos los modos de subida usan un único PUT por archivo y rechazan archivos de más de 5 GiB antes de enviar una solicitud. El streaming no incorpora subidas multiparte ni reanudación. Para archivos más grandes o partes reanudables, usa las subidas multiparte de R2.
Buenas prácticas de seguridad
Usa credenciales cuyo ámbito se limite al bucket y elige cuidadosamente las claves de destino. Este programa Scala del lado del servidor no necesita configuración CORS del navegador ni una URL pública del bucket. Para los archivos proporcionados por los usuarios, completa la validación antes de exportarlos a una ubicación desde la que tu aplicación los sirva.
Errores comunes y sus soluciones
| Resultado | Qué revisar a continuación |
|---|---|
| Error HTTP | Verifica el endpoint de la cuenta, el nombre del bucket y los permisos del token sobre los objetos. Un estado por sí solo no identifica qué ajuste es incorrecto. |
| Error local | Revisa las cuatro variables de entorno, las restricciones de los argumentos, que los archivos se puedan leer y el límite de 5 GiB. |
| Tiempo de espera agotado o error de conexión | Revisa la conectividad y los límites de las solicitudes; luego inspecciona el destino antes de volver a intentarlo. |
| Lote paralelo parcial | Inspecciona cada clave solicitada; los objetos de las subidas hermanas exitosas se conservan aunque haya fallado el comando en su conjunto. |
Conclusión
Usa el cuerpo basado en archivo del SDK para las exportaciones habituales y FS2 cuando tu pipeline de streaming existente necesite proporcionar los bytes. Si la exportación sigue a una tarea de procesamiento de Transloadit, el 🤖 /cloudflare/store Robot puede escribir los resultados en R2 en lugar de ejecutar este programa.