INTEGRITY Dokumentace

← R2 Data Catalog / r2-data-catalog / config-examples

Spark (Scala)

Níže je uveden příklad, jak můžete sestavit Apache Spark ↗ aplikaci (v jazyce Scala), která se připojuje k R2 Data Catalog. Tato aplikace je určena ke spuštění lokálně, ale lze ji upravit pro spuštění na clusteru.

Předpoklady

Příklad použití

Nejprve si na svém počítači vytvořte nový prázdný adresář projektu.

Uvnitř tohoto adresáře vytvořte následující soubor na cestě src/main/scala/com/example/R2DataCatalogDemo.scala. Ten bude sloužit jako hlavní vstupní bod vaší aplikace Spark.

package com.example

import org.apache.spark.sql.SparkSession

object R2DataCatalogDemo {
    def main(args: Array[String]): Unit = {

        val uri = sys.env("CATALOG_URI")
        val warehouse = sys.env("WAREHOUSE")
        val token = sys.env("TOKEN")

        val spark = SparkSession.builder()
            .appName("My R2 Data Catalog Demo")
            .master("local[*]")
            .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
            .config("spark.sql.catalog.mydemo", "org.apache.iceberg.spark.SparkCatalog")
            .config("spark.sql.catalog.mydemo.type", "rest")
            .config("spark.sql.catalog.mydemo.uri", uri)
            .config("spark.sql.catalog.mydemo.warehouse", warehouse)
            .config("spark.sql.catalog.mydemo.token", token)
            .getOrCreate()

        import spark.implicits._

        val data = Seq(
            (1, "Alice", 25),
            (2, "Bob", 30),
            (3, "Charlie", 35),
            (4, "Diana", 40)
        ).toDF("id", "name", "age")

        spark.sql("USE mydemo")

        spark.sql("CREATE NAMESPACE IF NOT EXISTS demoNamespace")

        data.writeTo("demoNamespace.demotable").createOrReplace()

        val readResult = spark.sql("SELECT * FROM demoNamespace.demotable WHERE age > 30")
        println("Records with age > 30:")
        readResult.show()
    }
}

Pro vytvoření této aplikace a správu závislostí použijeme sbt (“simple build tool”) ↗. Následuje ukázkový build.sbt soubor, který umístíte do kořenového adresáře projektu. Je nakonfigurován tak, aby vytvářel takzvaný „fat JAR“, který obsahuje všechny potřebné závislosti.

name := "R2DataCatalogDemo"

version := "1.0"

val sparkVersion = "3.5.3"
val icebergVersion = "1.8.1"

// You need to use binaries of Spark compiled with either 2.12 or 2.13; and 2.12 is more common.
// If you download Spark 3.5.3 with sdkman, then it comes with 2.12.18
scalaVersion := "2.12.18"

libraryDependencies ++= Seq(
    "org.apache.spark" %% "spark-core" % sparkVersion,
    "org.apache.spark" %% "spark-sql" % sparkVersion,
    "org.apache.iceberg" % "iceberg-core" % icebergVersion,
    "org.apache.iceberg" % "iceberg-spark-runtime-3.5_2.12" % icebergVersion,
    "org.apache.iceberg" % "iceberg-aws-bundle" % icebergVersion,
)

// build a fat JAR with all dependencies
assembly / assemblyMergeStrategy := {
    case PathList("META-INF", "services", xs @ _*) => MergeStrategy.concat
    case PathList("META-INF", xs @ _*) => MergeStrategy.discard
    case "reference.conf" => MergeStrategy.concat
    case "application.conf" => MergeStrategy.concat
    case x if x.endsWith(".properties") => MergeStrategy.first
    case x => MergeStrategy.first
}

// For Java  17 Compatibility
Compile / javacOptions ++= Seq("--release", "17")

Chcete-li povolit plugin sbt-assembly ↗ (slouží k sestavení fat JAR souborů), přidejte následující do nového souboru v project/assembly.sbt:

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "1.2.0")

Ujistěte se, že máte nainstalované Java, Spark a sbt a že jsou dostupné ve vašem shellu. Pokud používáte SDKMAN, můžete je nainstalovat podle následujícího postupu:

sdk install java 17.0.14-amzn
sdk install spark 3.5.3
sdk install sbt 1.10.11

Jakmile máte vše nainstalováno, můžete sestavit projekt pomocí sbt. Tím se vygeneruje jeden sloučený soubor JAR.

sbt clean assembly

Po sestavení by se výstupní soubor JAR měl nacházet v target/scala-2.12/R2DataCatalogDemo-assembly-1.0.jar.

Ke spuštění aplikace použijete spark-submit. Níže je ukázkový shellový skript (submit.sh) která obsahuje potřebné příznaky kompatibility Java pro Spark na Java 17:

# We need to set these "--add-opens" so that Spark can run on Java 17 (it needs access to
# parts of the JVM which have been modularized and made internal).
JAVA_17_COMPATIBILITY="--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"

spark-submit \
--conf "spark.driver.extraJavaOptions=$JAVA_17_COMPATIBILITY" \
--conf "spark.executor.extraJavaOptions=$JAVA_17_COMPATIBILITY" \
--class com.example.R2DataCatalogDemo target/scala-2.12/R2DataCatalogDemo-assembly-1.0.jar

Před spuštěním se ujistěte, že je skript spustitelný:

chmod +x submit.sh

V tomto okamžiku by měl váš adresář projektu vypadat takto:

Před odesláním úlohy se ujistěte, že máte nastavené požadované proměnné prostředí pro URI katalogu, warehouse a Token Cloudflare API.

export CATALOG_URI=
export WAREHOUSE=
export TOKEN=

Nyní jste připraveni spustit úlohu:

./submit.sh