|
| 1 | +--- |
| 2 | +title: Spark |
| 3 | +pcx_content_type: example |
| 4 | +--- |
| 5 | + |
| 6 | +Below is an example of how you can build an [Apache Spark](https://spark.apache.org/) application (with Scala) which connects to the R2 Data Catalog. This application is built to run locally, but it can be adapted to run on a cluster. |
| 7 | + |
| 8 | +## Prerequisites |
| 9 | + |
| 10 | +- Sign up for a [Cloudflare account](https://dash.cloudflare.com/sign-up/workers-and-pages). |
| 11 | +- Create an [R2 bucket](/r2/buckets/) and enable the data catalog. |
| 12 | +- Create an [R2 API token](/r2/api/tokens/) with both [R2 and data catalog permissions](/r2/api/tokens/#permissions). |
| 13 | +- Install Java 17, Spark 3.5.3, and SBT 1.10.11 |
| 14 | + - Note: The specific versions of tools are critical for getting things to work in this example. |
| 15 | + - Tip: [“SDKMAN”](https://sdkman.io/) is a convenient package manager for installing SDKs. |
| 16 | + |
| 17 | +## Example usage |
| 18 | + |
| 19 | +Create a new empty project directory somewhere on your machine. In your project directory, create the following file at `src/main/scala/com/example/R2DataCatalogDemo.scala`. |
| 20 | + |
| 21 | +```java |
| 22 | +package com.example |
| 23 | + |
| 24 | +import org.apache.spark.sql.SparkSession |
| 25 | + |
| 26 | +object R2DataCatalogDemo { |
| 27 | + def main(args: Array[String]): Unit = { |
| 28 | + |
| 29 | + val uri = sys.env("CATALOG_URI") |
| 30 | + val warehouse = sys.env("WAREHOUSE") |
| 31 | + val token = sys.env("TOKEN") |
| 32 | + |
| 33 | + val spark = SparkSession.builder() |
| 34 | + .appName("My R2 Data Catalog Demo") |
| 35 | + .master("local[*]") |
| 36 | + .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") |
| 37 | + .config("spark.sql.catalog.mydemo", "org.apache.iceberg.spark.SparkCatalog") |
| 38 | + .config("spark.sql.catalog.mydemo.type", "rest") |
| 39 | + .config("spark.sql.catalog.mydemo.uri", uri) |
| 40 | + .config("spark.sql.catalog.mydemo.warehouse", warehouse) |
| 41 | + .config("spark.sql.catalog.mydemo.token", token) |
| 42 | + .getOrCreate() |
| 43 | + |
| 44 | + import spark.implicits._ |
| 45 | + |
| 46 | + val data = Seq( |
| 47 | + (1, "Alice", 25), |
| 48 | + (2, "Bob", 30), |
| 49 | + (3, "Charlie", 35), |
| 50 | + (4, "Diana", 40) |
| 51 | + ).toDF("id", "name", "age") |
| 52 | + |
| 53 | + spark.sql("USE mydemo") |
| 54 | + |
| 55 | + spark.sql("CREATE NAMESPACE IF NOT EXISTS demoNamespace") |
| 56 | + |
| 57 | + data.writeTo("demoNamespace.demotable").createOrReplace() |
| 58 | + |
| 59 | + val readResult = spark.sql("SELECT * FROM demoNamespace.demotable WHERE age > 30") |
| 60 | + println("Records with age > 30:") |
| 61 | + readResult.show() |
| 62 | + } |
| 63 | +} |
| 64 | +``` |
| 65 | + |
| 66 | +For this demo, we will use [sbt (“simple build tool”)](https://www.scala-sbt.org/) to build the application and manage dependencies. Here is an example `built.sbt` file (stored in your project root) for running this application. It will produce a “fat JAR” which bundles the dependencies into a single JAR. |
| 67 | + |
| 68 | +```java |
| 69 | +name := "R2DataCatalogDemo" |
| 70 | + |
| 71 | +version := "1.0" |
| 72 | + |
| 73 | +val sparkVersion = "3.5.3" |
| 74 | +val icebergVersion = "1.8.1" |
| 75 | + |
| 76 | +// You need to use binaries of Spark compiled with either 2.12 or 2.13; and 2.12 is more common. |
| 77 | +// If you download Spark 3.5.3 with sdkman, then it comes with 2.12.18 |
| 78 | +scalaVersion := "2.12.18" |
| 79 | + |
| 80 | +libraryDependencies ++= Seq( |
| 81 | + "org.apache.spark" %% "spark-core" % sparkVersion, |
| 82 | + "org.apache.spark" %% "spark-sql" % sparkVersion, |
| 83 | + "org.apache.iceberg" % "iceberg-core" % icebergVersion, |
| 84 | + "org.apache.iceberg" % "iceberg-spark-runtime-3.5_2.12" % icebergVersion, |
| 85 | + "org.apache.iceberg" % "iceberg-aws-bundle" % icebergVersion, |
| 86 | +) |
| 87 | + |
| 88 | +// build a fat JAR with all dependencies |
| 89 | +assembly / assemblyMergeStrategy := { |
| 90 | + case PathList("META-INF", "services", xs @ _*) => MergeStrategy.concat |
| 91 | + case PathList("META-INF", xs @ _*) => MergeStrategy.discard |
| 92 | + case "reference.conf" => MergeStrategy.concat |
| 93 | + case "application.conf" => MergeStrategy.concat |
| 94 | + case x if x.endsWith(".properties") => MergeStrategy.first |
| 95 | + case x => MergeStrategy.first |
| 96 | +} |
| 97 | + |
| 98 | +// For Java 17 Compatability |
| 99 | +Compile / javacOptions ++= Seq("--release", "17") |
| 100 | +``` |
| 101 | + |
| 102 | +[Assembly](https://github.com/sbt/sbt-assembly?tab=readme-ov-file) is a plugin that can build fat JARs. Create a `project/assembly.sbt` and add the following: |
| 103 | + |
| 104 | +``` |
| 105 | +addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "1.2.0") |
| 106 | +``` |
| 107 | + |
| 108 | +You can build the project by running sbt. Be sure that you have installed Java, Spark, and SBT. You may need to restart your shell. |
| 109 | + |
| 110 | +```bash |
| 111 | +sdk install java 17.0.14-amzn |
| 112 | +sdk install spark 3.5.3 |
| 113 | +sdk install sbt 1.10.11 |
| 114 | +``` |
| 115 | + |
| 116 | +Now you can run sbt to build the fat JAR. |
| 117 | + |
| 118 | +```bash |
| 119 | +sbt clean assembly |
| 120 | +``` |
| 121 | + |
| 122 | +The fat JAR should be available at `target/scala-2.12/R2DataCatalogDemo-assembly-1.0.jar`. Now you can run via `spark-submit`. Here's an example shell script at `submit.sh` to execute it: |
| 123 | + |
| 124 | +``` |
| 125 | +# We need to set these "--add-opens" so that Spark can run on Java 17 (it needs access to |
| 126 | +# parts of the JVM which have been modularized and made internal). |
| 127 | +JAVA_17_COMPATABILITY="--add-opens=java.base/sun.nio.ch=ALL-UNNAMED --add-opens=java.base/java.nio=ALL-UNNAMED --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED" |
| 128 | +
|
| 129 | +spark-submit \ |
| 130 | +--conf "spark.driver.extraJavaOptions=$JAVA_17_COMPATABILITY" \ |
| 131 | +--conf "spark.executor.extraJavaOptions=$JAVA_17_COMPATABILITY" \ |
| 132 | +--class com.example.R2DataCatalogDemo target/scala-2.12/R2DataCatalogDemo-assembly-1.0.jar |
| 133 | +``` |
| 134 | + |
| 135 | +To make your script executable run: |
| 136 | + |
| 137 | +```bash |
| 138 | +chmod +x submit.sh |
| 139 | +``` |
| 140 | + |
| 141 | +Just to recap, your project structure should look something like this: |
| 142 | + |
| 143 | +``` |
| 144 | +. |
| 145 | +├── Makefile |
| 146 | +├── README.md |
| 147 | +├── build.sbt |
| 148 | +├── project |
| 149 | +│ ├── assembly.sbt |
| 150 | +│ ├── build.properties |
| 151 | +│ └── project |
| 152 | +├── spark-submit.sh |
| 153 | +└── src |
| 154 | + └── main |
| 155 | + └── scala |
| 156 | + └── com |
| 157 | + └── example |
| 158 | + └── R2DataCatalogDemo.scala |
| 159 | +``` |
| 160 | + |
| 161 | +To run, set environment variables for your catalog's URI, warehouse name, and your Cloudflare API token. |
| 162 | + |
| 163 | +```bash |
| 164 | +export CATALOG_URI= |
| 165 | +export WAREHOUSE= |
| 166 | +export TOKEN= |
| 167 | +``` |
| 168 | + |
| 169 | +Now you may run your submit script: |
| 170 | + |
| 171 | +```bash |
| 172 | +./submit.sh |
| 173 | +``` |
0 commit comments