Gunakan cluster gunung es dengan Spark

Dimulai dengan Amazon EMR versi 6.5.0, Anda dapat menggunakan Iceberg dengan cluster Spark Anda tanpa perlu menyertakan tindakan bootstrap. Untuk Amazon EMR versi 6.4.0 dan sebelumnya, Anda dapat menggunakan tindakan bootstrap untuk melakukan pra-instal semua dependensi yang diperlukan.

Dalam tutorial ini, Anda menggunakan AWS CLI untuk bekerja dengan Iceberg di cluster Amazon EMR Spark. Untuk menggunakan konsol untuk membuat cluster dengan Iceberg diinstal, ikuti langkah-langkah di Mem bangun danau data Apache Iceberg menggunakan Amazon Athena, Amazon EMR, dan Glue. AWS

Buat cluster gunung es

Anda dapat membuat cluster dengan Iceberg diinstal menggunakan Konsol Manajemen AWS, AWS CLI atau Amazon EMR API. Dalam tutorial ini, Anda menggunakan AWS CLI untuk bekerja dengan Iceberg di cluster Amazon EMR. Untuk menggunakan konsol untuk membuat cluster dengan Iceberg diinstal, ikuti langkah-langkah di Mem bangun danau data Apache Iceberg menggunakan Amazon Athena, Amazon EMR, dan Glue. AWS

Untuk menggunakan Iceberg di Amazon EMR dengan AWS CLI, pertama buat cluster dengan langkah-langkah berikut. Untuk informasi tentang menentukan klasifikasi Gunung Es menggunakan AWS CLI, lihat Menyediakan konfigurasi menggunakan AWS CLI saat Anda membuat klaster atauSediakan konfigurasi menggunakan Java SDK ketika Anda membuat sebuah klaster.

1. Buat configurations.json file dengan konten berikut:

   [{
       "Classification":"iceberg-defaults",
       "Properties":{"iceberg.enabled":"true"}
   }]

2. Selanjutnya, buat cluster dengan konfigurasi berikut. Ganti contoh jalur bucket Amazon S3 dan ID subnet dengan milik Anda sendiri.

   aws emr create-cluster --release-label emr-6.5.0 \
   --applications Name=Spark \
   --configurations file://configurations.json \
   --region us-east-1 \
   --name My_Spark_Iceberg_Cluster \
   --log-uri s3://amzn-s3-demo-bucket/ \
   --instance-type m5.xlarge \
   --instance-count 2 \
   --service-role EMR_DefaultRole_V2 \ 
   --ec2-attributes InstanceProfile=EMR_EC2_DefaultRole,SubnetId=subnet-1234567890abcdef0

Atau, Anda dapat membuat cluster Amazon EMR termasuk aplikasi Spark dan menyertakan file /usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar sebagai dependensi JAR dalam pekerjaan Spark. Untuk informasi selengkapnya, lihat Men girimkan Aplikasi.

Untuk menyertakan jar sebagai dependensi dalam pekerjaan Spark, tambahkan properti konfigurasi berikut ke aplikasi Spark:

--conf "spark.jars=/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar"

Untuk informasi selengkapnya tentang dependensi pekerjaan Spark, lihat Manajemen Keter gantungan di dokumen Apache Spark Menjalankan Spark di Ku bernetes.

Menginisialisasi sesi Spark untuk Iceberg

Contoh berikut menunjukkan cara meluncurkan shell Spark interaktif, menggunakan Spark submit, atau menggunakan Notebook Amazon EMR untuk bekerja dengan Iceberg di Amazon EMR.

spark-shell

1. Connect ke simpul utama menggunakan SSH. Untuk informasi selengkapnya, lihat Connect ke simpul utama menggunakan SSH di Panduan Pengelolaan Amazon EMR.

2. Masukkan perintah berikut untuk meluncurkan shell Spark. Untuk menggunakan cang PySpark kang, ganti spark-shell denganpyspark.

   spark-shell \
       --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
       --conf spark.sql.catalog.my_catalog.warehouse=s3://amzn-s3-demo-bucket/prefix/
       --conf spark.sql.catalog.my_catalog.type=glue \
       --conf spark.sql.defaultCatalog=my_catalog \
       --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

spark-submit

1. Connect ke simpul utama menggunakan SSH. Untuk informasi selengkapnya, lihat Connect ke simpul utama menggunakan SSH di Panduan Pengelolaan Amazon EMR.

2. Masukkan perintah berikut untuk meluncurkan sesi Spark untuk Iceberg.

   spark-submit \
   --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
   --conf spark.sql.catalog.my_catalog.warehouse=s3://amzn-s3-demo-bucket1/prefix \
   --conf spark.sql.catalog.my_catalog.type=glue \
   --conf spark.sql.defaultCatalog=my_catalog \
   --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

EMR Studio notebooks

Untuk menginisialisasi sesi Spark menggunakan buku catatan EMR Studio, konfigurasikan sesi Spark Anda menggunakan perintah %%configure ajaib di notebook Amazon EMR Anda, seperti pada contoh berikut. Untuk informasi selengkapnya, lihat Menggunakan sihir Notebook EMR di Panduan Manajemen Amazon EMR.

%%configure -f{
"conf":{
    "spark.sql.catalog.my_catalog":"org.apache.iceberg.spark.SparkCatalog",
    "spark.sql.catalog.my_catalog.type":"glue",
    "spark.sql.catalog.my_catalog.warehouse":"s3://amzn-s3-demo-bucket1/prefix/",
    "spark.sql.defaultCatalog":"my_catalog",
    "spark.sql.extensions":"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
    }
}

CLI

Untuk menginisialisasi cluster Spark menggunakan CLI dan mengatur semua konfigurasi default sesi Spark Iceberg, jalankan contoh berikut. Untuk informasi selengkapnya tentang menentukan klasifikasi konfigurasi menggunakan AWS CLI dan Amazon EMR API, lihat Meng onfigurasi aplikasi.

[
  {
    "Classification": "spark-defaults",
    "Properties": {
      "spark.sql.catalog.my_catalog":"org.apache.iceberg.spark.SparkCatalog",
      "spark.sql.catalog.my_catalog.type":"glue",
      "spark.sql.catalog.my_catalog.warehouse":"s3://amzn-s3-demo-bucket1/prefix/",
      "spark.sql.defaultCatalog":"my_catalog",
      "spark.sql.extensions":"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
    }
  }
]

Menulis ke meja gunung es

Contoh berikut menunjukkan cara membuat DataFrame dan menulisnya sebagai dataset gunung es. Contoh-contoh menunjukkan bekerja dengan kumpulan data menggunakan shell Spark saat terhubung ke node master menggunakan SSH sebagai pengguna hadoop default.

catatan

Untuk menempelkan sampel kode ke dalam shell Spark, ketik :paste pada prompt, tempel contoh, lalu tekanCTRL+D.

PySpark

Spark menyertakan Python-based shell,pyspark, yang dapat Anda gunakan untuk membuat prototipe program Spark yang ditulis dalam Python. Memang pyspark gil pada node master.

## Create a DataFrame.
data = spark.createDataFrame([
 ("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
 ("101", "2015-01-01", "2015-01-01T12:14:58.597216Z"),
 ("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
 ("103", "2015-01-01", "2015-01-01T13:51:40.519832Z")
],["id", "creation_date", "last_update_time"])

## Write a DataFrame as a Iceberg dataset to the Amazon S3 location.
spark.sql("""CREATE TABLE IF NOT EXISTS dev.db.iceberg_table (id string,
creation_date string,
last_update_time string)
USING iceberg
location 's3://amzn-s3-demo-bucket/example-prefix/db/iceberg_table'""")

data.writeTo("dev.db.iceberg_table").append()

Scala

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.functions._

// Create a DataFrame.
val data = Seq(
("100", "2015-01-01", "2015-01-01T13:51:39.340396Z"),
("101", "2015-01-01", "2015-01-01T12:14:58.597216Z"),
("102", "2015-01-01", "2015-01-01T13:51:40.417052Z"),
("103", "2015-01-01", "2015-01-01T13:51:40.519832Z")
).toDF("id", "creation_date", "last_update_time")

// Write a DataFrame as a Iceberg dataset to the Amazon S3 location.
spark.sql("""CREATE TABLE IF NOT EXISTS dev.db.iceberg_table (id string,
creation_date string,
last_update_time string)
USING iceberg
location 's3://amzn-s3-demo-bucket/example-prefix/db/iceberg_table'""")

data.writeTo("dev.db.iceberg_table").append()

Baca dari meja gunung es

PySpark

df = spark.read.format("iceberg").load("dev.db.iceberg_table")
df.show()

Scala

val df = spark.read.format("iceberg").load("dev.db.iceberg_table")
df.show()

Spark SQL

SELECT * from dev.db.iceberg_table LIMIT 10

Penggunaan AWS Katalog Data Lem dengan Spark Iceberg

Anda dapat terhubung ke Katalog Data AWS Glue dari Spark Iceberg. Bagian ini menunjukkan perintah yang berbeda untuk menghubungkan.

Hubungkan ke default AWS Katalog lem di wilayah default Anda

Contoh ini menunjukkan cara menghubungkan, menggunakan jenis katalog Glue. Jika Anda tidak menentukan ID katalog, itu menggunakan default:

spark-submit \
    --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.my_catalog.warehouse=s3://amzn-s3-demo-bucket1/prefix \
    --conf spark.sql.catalog.my_catalog.type=glue \
    --conf spark.sql.defaultCatalog=my_catalog \
    --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Hubungkan ke AWS Katalog lem dengan ID katalog tertentu

Contoh ini menunjukkan cara menghubungkan, menggunakan ID katalog:

spark-submit \
    --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.my_catalog.warehouse=s3://amzn-s3-demo-bucket1/prefix \
    --conf spark.sql.catalog.my_catalog.type=glue \
    --conf spark.sql.catalog.my_catalog.glue.id=AWS Glue catalog ID \
    --conf spark.sql.defaultCatalog=my_catalog \
    --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Perintah ini dapat digunakan untuk menghubungkan ke katalog AWS Glue di akun yang berbeda, atau ke katalog RMS, atau ke katalog federasi.

Menggunakan Katalog REST Iceberg (IRC) dengan Spark Iceberg

Bagian berikut merinci cara mengonfigurasi integrasi Iceberg dengan katalog.

Sambungkan ke AWS Titik akhir IRC Katalog Data Lem

Berikut ini menunjukkan contoh spark-submit perintah untuk menggunakan Iceberg REST:

spark-submit \
    --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.my_catalog.warehouse=glue catalog ID \
    --conf spark.sql.catalog.my_catalog.type=rest \
    --conf spark.sql.catalog.my_catalog.uri=glue endpoint URI/iceberg \
    --conf spark.sql.catalog.my_catalog.rest.sigv4-enabled=true \
    --conf spark.sql.catalog.my_catalog.rest.signing-name=glue \
    --conf spark.sql.defaultCatalog=my_catalog \
    --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Untuk menggunakannya pada cluster yang diaktifkan runtime-role, pengaturan konfigurasi percikan tambahan berikut diperlukan:

"spark.hadoop.fs.s3.credentialsResolverClass": "software.amazon.glue.GlueTableCredentialsResolver",
"spark.hadoop.catalog-impl": "org.apache.iceberg.aws.glue.GlueCatalog",
"spark.hadoop.glue.id": glue catalog ID
"spark.hadoop.glue.endpoint": "glue endpoint"

Untuk daftar URL titik akhir AWS Glue untuk setiap wilayah, lihat Titik akhir dan kuota AWS Glue.

Sambungkan ke titik akhir IRC yang sewenang-wenang

Berikut ini menunjukkan contoh spark-submit perintah untuk menggunakan titik akhir IRC:

spark-submit \
    --conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
    --conf spark.sql.catalog.my_catalog.warehouse=warehouse name \
    --conf spark.sql.catalog.my_catalog.type=rest \
    --conf spark.sql.catalog.my_catalog.uri=your rest endpoint \
    --conf spark.sql.defaultCatalog=my_catalog \
    --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Perbedaan konfigurasi saat Anda menggunakan Iceberg versus SparkCatalog SparkSessionCatalog

Iceberg menyediakan dua cara untuk membuat katalog Spark Iceberg. Anda dapat mengatur konfigurasi Spark ke salah satu SparkCatalog atau keSparkSessionCatalog.

Menggunakan Iceberg SparkCatalog

Berikut ini menunjukkan perintah untuk digunakan SparkCatalog sebagai katalog Spark Iceberg:

spark-shell \
--conf spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.my_catalog.warehouse=s3://amzn-s3-demo-bucket1/prefix \
--conf spark.sql.catalog.my_catalog.type=glue \
--conf spark.sql.defaultCatalog=my_catalog

Pertimbangan untuk pendekatan ini:

• Anda dapat mengakses tabel Iceberg tetapi tidak ada tabel lain.

• Nama katalog tidak dapat berupa spark_ catalog. Ini adalah nama katalog awal di Spark. Itu selalu terhubung ke metastore Hive. Ini adalah katalog default di Spark, kecuali pengguna menulisnya menggunakanspark.sql.defaultCatalog.

• Anda dapat menyet spark.sql.defaultCatalog el ke nama katalog Anda untuk menjadikannya katalog default.

Menggunakan Iceberg SparkSessionCatalog

Berikut ini menunjukkan perintah untuk digunakan SparkSessionCatalog sebagai katalog Spark Iceberg:

spark-shell \
    --conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \
    --conf spark.sql.catalog.spark_catalog.warehouse=s3://amzn-s3-demo-bucket1/prefix \
    --conf spark.sql.catalog.spark_catalog.type=glue

Pertimbangan untuk pendekatan ini:

• Jika tabel tidak ditemukan sebagai tabel Iceberg, Spark akan mencoba melihat apakah itu tabel di metastore Hive. Lihat Menggunakan Katalog Data AWS Lem sebagai katalog untuk Hive untuk informasi selengkapnya.

• Nama katalog harus spark _catalog.

Menggunakan ekstensi Iceberg Spark

Iceberg menawarkan ekstensi Spark org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions yang dapat diatur pengguna melalui konfigurasi spark.sql.extensions ekstensi Spark. Ekstensi mengaktifkan fitur Iceberg utama seperti DELETE level baris, UPDATE dan MERGE, pernyataan dan prosedur bahasa definisi data Iceberg-specific Spark, seperti pemadatan, kedaluwarsa snapshot, percabangan dan penandaan, dan sebagainya. Lihat berikut ini untuk lebih jelasnya:

• Ekstensi tulis Iceberg Spark: Spark Writes

• Ekstensi DDL Iceberg Spark: Ekstensi SQL ALTER TABLE

• Perpanjangan prosedur Iceberg Spark: Prosedur Spark

Pertimbangan untuk menggunakan Iceberg dengan Spark

• Amazon EMR 6.5.0 tidak mendukung Iceberg yang berjalan di Amazon EMR di EKS secara default. Gambar kustom Amazon EMR 6.5.0 tersedia sehingga Anda dapat meneruskannya --jars local:///usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar sebagai spark-submit parameter untuk membuat tabel Iceberg di Amazon EMR di EKS. Untuk informasi selengkapnya, lihat Mengirimkan beban kerja Spark di Amazon EMR menggunakan gambar khusus di Panduan Peng embangan Amazon EMR di EKS. Anda juga dapat menghubungi Dukungan untuk bantuan. Dimulai dengan Amazon EMR 6.6.0, Iceberg didukung di Amazon EMR di EKS.

• Saat menggunakan AWS Glue sebagai katalog untuk Iceberg, pastikan database tempat Anda membuat tabel ada di AWS Glue. Jika Anda menggunakan layanan seperti AWS Lake Formation dan Anda tidak dapat memuat katalog, pastikan Anda memiliki akses yang tepat ke layanan untuk menjalankan perintah.

• Jika Anda menggunakan Iceberg SparkSessionCatalog, seperti yang dijelaskan dalamPerbedaan konfigurasi saat Anda menggunakan Iceberg versus SparkCatalog SparkSessionCatalog, Anda harus mengikuti langkah-langkah konfigurasi yang dijelaskan dalam Konfigurasi Katalog Data AWS Glue sebagai metastore Apache Hive, selain mengonfigurasi pengaturan Katalog Data Spark Iceberg AWS Glue.