Vai al contenuto principale

Tavoli Ilum

Ilum Tables è un formato Spark che funge da wrapper per i formati di dati Delta, Iceberg e Hudi. Consente di accedere e creare set di dati in questi formati utilizzando un'interfaccia unificata, con conseguente progettazione del codice più flessibile.

Importazione di tabelle Ilum​

Per utilizzare Tavoli Ilum Dovresti includere formato ilum-spark pacchetto nel tuo Offerte di lavoro per Ilum. A tale scopo, è possibile aggiungere questa configurazione:

spark.jars.packages=cloud.ilum:ilum-spark-format:6.1.0

o aggiungendo il pacchetto come jar separato nelle risorse del tuo Ilum Job

Come includere il pacchetto nel tuo Applicazione Scala

  • Utilizzo di sbt:
libraryDependencies += "cloud.ilum" % "ilum-spark-format" % "6.1.0"
  • Utilizzo di Maven:
<dipendenza>
    <groupId>cloud.ilumgroupId>
    <artifactId>formato ilum-sparkartifactId>
    <Versione>6.1.0Versione>
dipendenza>
  • Utilizzo di gradle:
Gruppo di implementazione: 'cloud.ilum', nome: 'ilum-spark-format', versione: '6.1.0'

Come usarlo?​

  • Lettura e scrittura dei dati specificando 'formato ilum'
  • Leggere e scrivere dati utilizzando Metodo ILUM

Per fare questo è necessario importarli in questo modo:

importazione nuvola.ilum.implicits.{
    IlumDataFrameReader, 
    IlumDataFrameWriter, 
    IlumDataFrameWriterV2, 
    IlumDataStreamWriter, 
    IlumDataStreamReader
}
  • Leggere e scrivere dati preconfigurando il catalogo ( writeTo, read.table)

Lettura​

val filepath = "s3a://ilum-files/ilum-tables/table"
val tableFormat = Some("delta")

// without ilum method
val mydf = scintilla.leggere.format("ilum").opzione("tableFormat", tableFormat).load(filepath)

// with ilum method
val mydf2 = sparkSession.leggere.ilum(filePath, tableFormat)


Scrittura​

val filepath = "s3a://ilum-files/ilum-tables/table"
val tableFormat = "delta"

val dati = Seq(
  (1, "Alice"),
  (2, "Bob"),
  (3, "Cathy")
)

val Df = scintilla.createDataFrame(dati).toDF("id", "nome")

// using DataframeWriterV1

// you can use syntax like this
Df.scrivere.format("ilum").opzione("tableFormat", tableFormat).save(filepath)

// or you can use ilum function

Df.scrivere.ilum(filepath + "/1", format)

// using DataframeWriterV2 with preconfigured Delta catalog
val catalogo = "catalog"
val tavolo = "tablename"
Df.writeTo(s"${catalogo}.${tavolo}").ilum(format, None ).createOrReplace()

Streaming​

Senza metodi Ilum

val filepath = "s3a://ilum-files/ilum-tables/streaming"
val tableFormat = "delta"

val input = scintilla.readStream
    .format("ilum")
    .opzione("tableFormat", tableFormat)
    .load(filepath)

val quesito = input.writeStream 
.outputMode("append") 
    .format("ilum")
    .opzione("tableFormat", tableFormat)
    .opzione("path", filepath + "_copy")
    .opzione("checkpointLocation", filepath + "_checkpoint")
    .start()

quesito.awaitTermination()

Con i metodi Ilum:


val filePath = s"s3a://ilum-files/ilum-tables/smth"
val tableFormat = Some("delta")

val Df = sparkSession.readStream.ilum(filePath, tableFormat)

val quesito = Df.writeStream
    .opzione("checkpointLocation", filePath + "_checkpoint")
    .ilum(filePath+"_copy", tableFormat)

quesito.awaitTermination()

Configurazione dei formati dei dati​

Delta​

Per utilizzare Delta, è necessario utilizzare queste configurazioni di spark:

spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension
spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog
spark.sql.warehouse.dir=s3a://ilum-files/ilum-warehouse

ed è necessario includere il pacchetto Delta nell'ambiente. A tale scopo, è possibile utilizzare l'immagine spark di Kubernetes con le estensioni delta preinstallato:

spark.kubernetes.container.image=ilum/spark:3.5.2-delta

o installare autonomamente il pacchetto di estensione richiesto

Iceberg​

Per utilizzare Iceberg è necessario aggiungere queste configurazioni:

spark.sql.catalog.iceberg_catalog=org.apache.iceberg.spark.SparkSessionCatalog
spark.sql.catalog.iceberg_catalog.type=Alveare
spark.sql.catalog.iceberg_catalog, org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.iceberg_catalog.type=Hadoop
spark.sql.catalog.iceberg_catalog.warehouse=s3a://ilum-files/ilum-tables/iceberg/warehouse

e devi includere org.apache.iceberg:iceberg-spark-runtime-3.5_2.13:1.6.1 Pacchetto nell'ambiente aggiungendo il relativo file JAR alle risorse o aggiungendolo a configurazioni Spark come questo:

spark.jars.packages=org.apache.iceberg:iceberg-spark-runtime-3.5_2.13:1.6.1

Hudi​

Per utilizzare Hudi è necessario aggiungere queste configurazioni:

spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSpearkSessionExtension
spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog

e importa org.apache.hudi:hudi-spark3.5-bundle_2.12:0.15.0 Pacchetto nell'ambiente aggiungendo il file JAR alle risorse o aggiungendolo alle configurazioni Spark in questo modo:

spark.jars.packages=org.apache.hudi:hudi-spark3.5-bundle_2.12:0.15.0
Last updated on