Dateien mit Scala und AWS SDK zu Cloudflare R2 exportieren

Kevin van Zonneveld

Mitgründer · Amsterdam, Niederlande ·

Exportieren Sie fertige lokale Dateien aus einer Scala-Anwendung mit dem AWS SDK for Java v2 in einen privaten Cloudflare-R2-Bucket. Dieses Beispiel bietet ein Kommandozeilenprogramm mit synchronen, asynchronen, FS2-Streaming- und parallelen Upload-Modi sowie eine Möglichkeit, die gespeicherten Bytes zu überprüfen.

Warum Cloudflare R2 wählen?

Die S3-kompatible API von R2 ermöglicht es Scala-Code, die Anfragesignierung und HTTP-Clients des Java-SDKs zu nutzen. Das Beispiel verwendet einen Bucket ohne Einschränkung auf einen Rechtsraum am standardmäßigen Konto-Endpunkt. Buckets für bestimmte Rechtsräume benötigen den entsprechenden Endpunkt.

Ihr Scala-Projekt einrichten

Diese Anleitung richtet sich an bestehende Scala-2.13-Anwendungen mit sbt 1.x. Die getesteten, festgelegten Versionen sind Scala 2.13.18, sbt 1.13.0, Cats Effect 3.7.1, FS2 3.14.0 und AWS SDK 2.56.0. Scala 3 und sbt 2 sind separate Migrationsentscheidungen; diese Versionsbindungen halten das Beispiel innerhalb der Buildfamilie von Scala 2.13.

Verwenden Sie Bash, cURL und eine gepflegte OpenJDK-21-Installation mit java im PATH. Die getestete JDK-Patchversion ist 21.0.12.1; verwenden Sie das aktuelle Sicherheitsupdate Ihres JDK-Anbieters innerhalb von 21. Sie benötigen weder einen separat installierten Scala-Compiler noch einen sbt-Befehl. Führen Sie Folgendes in einem beschreibbaren Verzeichnis aus, um ein neues Projekt zu erstellen; die Subshell lässt Ihr aktuelles Verzeichnis unverändert:

(
  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
)

Wenn die Erstellung fehlschlägt, weil r2-scala bereits existiert, wählen Sie ein anderes übergeordnetes Verzeichnis. Falls der Download fehlschlägt, behalten Sie das Projekt und wiederholen Sie darin nur den cURL-Befehl. Speichern Sie die folgenden Dateien in r2-scala. Verwenden Sie project/build.properties, um die Launcher-Version festzulegen:

sbt.version=1.13.0

Speichern Sie 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

Den AWS-SDK-Client konfigurieren

Erstellen Sie einen Bucket und darauf beschränkte R2-S3-Zugangsdaten mit Lese- und Schreibberechtigung für Objekte. Stellen Sie R2_ACCOUNT_ID, R2_ACCESS_KEY_ID, R2_SECRET_ACCESS_KEY und R2_BUCKET vor dem Programmstart über die Prozessumgebung bereit. Nutzen Sie einen Secret-Manager oder die Umgebungskonfiguration Ihres Terminals; hinterlegen Sie keine Zugangsdaten in der Scala-Datei oder in Befehlsargumenten. Der Bucket bleibt privat und muss bereits existieren.

Die Clients verwenden die Region auto, pfadbasierte Adressierung und eine deaktivierte Chunked-Kodierung des Anfragekörpers gemäß Cloudflares Java-Konfiguration. WHEN_REQUIRED deaktiviert optionale SDK-Anfrageprüfsummen. Das Programm wählt Apache für synchrones HTTP und Netty für asynchrones HTTP und schließt jeden Client mit Resource von Cats Effect.

Einfacher synchroner Upload

Wählen Sie sync für einen einfachen blockierenden PUT. Dabei wird die Datei mit RequestBody.fromFile im Pool für blockierende Aufrufe von Cats Effect gelesen. Ein Abbruch wartet darauf, dass der blockierende Aufruf zurückkehrt oder sein Zeitlimit erreicht; wählen Sie async oder fs2, wenn Sie das zugrunde liegende Future abbrechen können müssen.

Asynchroner Upload mit Cats Effect

Speichern Sie dieses vollständige Programm als src/main/scala/R2Upload.scala. async verwendet den dateibasierten Anfragekörper des SDKs; fs2 stellt einen Reactive-Streams-Publisher bereit. Beide warten über die CompletableFuture-Brücke von Cats Effect, die den Abbruch des Futures anfordert, wenn ihre Fiber abgebrochen wird.

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)
    }
  }
}

Öffnen Sie ein Terminal in r2-scala. Diese Befehle erstellen eine kleine Eingabedatei, kompilieren das Programm und laden sie hoch. Bei der ersten Kompilierung werden Abhängigkeiten heruntergeladen. Datei- und Schlüsselargumente dieses Launchers sind auf ASCII-Buchstaben, Ziffern, _, ., / und - beschränkt; setzen Sie ./ vor einen lokalen Namen, der mit einem Bindestrich beginnt. Zielschlüssel dürfen nicht mit - beginnen und müssen innerhalb eines Batches eindeutig sein. Lassen Sie die Quelldateien unverändert, bis alle Versuche abgeschlossen sind.

(
  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"
)

Ersetzen Sie async durch sync oder fs2, um die anderen Einzeldatei-Modi zu verwenden. Ein erfolgreicher PUT ersetzt jedes Objekt unter demselben Schlüssel, auch bei einer leeren Eingabedatei. Verwenden Sie einen neuen Schlüssel, wenn Sie das bisherige Objekt behalten möchten.

Lesen Sie nach erfolgreichem Upload das private Objekt wieder aus und vergleichen Sie die SHA-256-Hashes der vollständigen lokalen und gespeicherten Inhalte:

java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload verify ./hello.txt uploads/hello.txt"

Das Prüfprogramm gibt Stored bytes match. nur aus, wenn beide Hashes übereinstimmen, und liefert bei einer Abweichung oder einem fehlgeschlagenen Lesezugriff einen Status ungleich null. Ein abgeschlossener PUT allein bestätigt die Annahme durch den Dienst; führen Sie den Vergleich aus, bevor Sie ein wichtiges Original löschen. Auch für die Überprüfung muss die Quelle unverändert bleiben.

Antworten und Fehler behandeln

Upload completed. erscheint erst, wenn alle ausgewählten Uploads abgeschlossen sind und der Gültigkeitsbereich des Clients beendet ist. Dienstfehler liefern einen Exit-Code ungleich null und einen HTTP-Status. Lokale Konfigurations-, Datei- und Netzwerkfehler liefern ebenfalls einen Exit-Code ungleich null, ohne Zugangsdaten oder unverarbeitete SDK-Antworten auszugeben.

Ein abgebrochener Upload gilt nicht als erfolgreicher Abschluss. Ein Abbruch oder eine verlorene Antwort kann dennoch ein bereits gespeichertes Objekt am Ziel hinterlassen: Prüfen Sie es vor einem erneuten Versuch. Dieses Programm macht bei Fehlern keine Objektspeicherung rückgängig und löscht keine entfernten Daten.

Wiederholungslogik mit exponentiellem Backoff

Die Standardstrategie für Wiederholungsversuche des SDKs behandelt Fehler, bei denen ein erneuter Versuch möglich ist, mit höchstens drei Versuchen pro PUT. Das Zeitlimit für einen API-Aufruf beträgt einschließlich Wiederholungsversuchen zwei Minuten; das Zeitlimit pro Versuch beträgt 45 Sekunden. Für den Verbindungsaufbau gilt ein Limit von 10 Sekunden. Dies sind Anfragelimits, keine Frist von zwei Minuten für einen gesamten parallelen Batch. Die Leselimits der Transportschicht von 30 Sekunden liegen unter dem Zeitlimit pro Versuch; Netty hat außerdem ein Schreiblimit von 30 Sekunden.

Bei einer langsamen Übertragung kann das Zeitlimit auch unterhalb der maximalen Dateigröße überschritten werden. Prüfen Sie das Ziel und passen Sie dann die Limits an Ihre Verbindung an, bevor Sie die unveränderte Datei erneut hochladen. Fügen Sie keine unbegrenzte Wiederholungsschleife hinzu. Der FS2-Publisher kann seinen Dateistream bei einem neuen Abonnement erneut ausführen und benötigt daher ebenfalls eine unveränderte, erneut lesbare Quelle.

Viele Dateien parallel hochladen

Der Modus parallel führt höchstens vier Upload-Effekte gleichzeitig aus. Schlägt einer fehl, bricht er die übrigen Effekte ab und liefert einen Fehlerstatus. Bereits gespeicherte Objekte bleiben erhalten; ein Batch ist keine Transaktion. Nachdem Sie eine zweite lokale Datei namens report.pdf erstellt haben, führen Sie Folgendes aus:

java -Dsbt.server.autostart=false -jar sbt-launch.jar "runMain R2Upload parallel ./hello.txt uploads/hello.txt ./report.pdf uploads/report.pdf"

Das Limit gilt für aktive Uploads, nicht für die Argumentliste im Arbeitsspeicher oder jeden internen Puffer von Netty und dem SDK. Es teilt eine Datei nicht in parallele Teile auf.

Große Dateien mit fs2 streamen

fs2 liest Blöcke von 64 KiB und stellt sie je nach nachgelagertem Bedarf als Elemente des Typs ByteBuffer bereit. Die Ressource des Publishers bleibt bis zum Abschluss, Fehler oder Abbruch geöffnet. Dadurch wird vermieden, die gesamte Datei in ein Byte-Array zu laden; dies ist keine feste Obergrenze für den gesamten Speicherbedarf der JVM oder des HTTP-Clients.

Alle Upload-Modi verwenden einzelne PUTs und lehnen Dateien über 5 GiB ab, bevor eine Anfrage gesendet wird. Streaming bietet weder mehrteilige Uploads noch eine Wiederaufnahme. Verwenden Sie für größere Dateien oder fortsetzbare Teile mehrteilige R2-Uploads.

Bewährte Verfahren für Sicherheit

Verwenden Sie auf den Bucket beschränkte Zugangsdaten und wählen Sie Zielschlüssel bewusst. Dieses serverseitige Scala-Programm benötigt weder eine CORS-Konfiguration für den Browser noch eine öffentliche Bucket-URL. Schließen Sie bei Dateien, die Nutzer bereitstellen, die Validierung ab, bevor Sie sie an einen Speicherort exportieren, aus dem Ihre Anwendung Dateien ausliefert.

Häufige Fehler und ihre Behebung

Ergebnis	Nächster Prüfschritt
HTTP-Fehler	Prüfen Sie den Konto-Endpunkt, den Bucket-Namen und die Objektberechtigungen des Tokens. Ein Status allein zeigt nicht, welche Einstellung falsch ist.
Lokaler Fehler	Prüfen Sie alle vier Umgebungsvariablen, die Argumentbeschränkungen, die Lesbarkeit der Dateien und das Limit von 5 GiB.
Zeitüberschreitung oder Verbindungsfehler	Prüfen Sie die Verbindung und die Anfragelimits, dann das Ziel, bevor Sie es erneut versuchen.
Teilweise abgeschlossener paralleler Batch	Prüfen Sie jeden angeforderten Schlüssel; erfolgreich gespeicherte Objekte bleiben erhalten, obwohl der gesamte Befehl fehlgeschlagen ist.

Fazit

Verwenden Sie für gewöhnliche Exporte den dateibasierten Anfragekörper des SDKs und FS2, wenn Ihre bestehende Streaming-Pipeline die Bytes bereitstellen muss. Folgt der Export auf einen Verarbeitungsauftrag bei Transloadit, kann der Robot 🤖 /cloudflare/store die Ergebnisse in R2 speichern, statt dieses Programm auszuführen.