Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Gunakan cluster Iceberg dengan Spark

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

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

Buat cluster Iceberg

Anda dapat membuat cluster dengan Iceberg diinstal menggunakan AWS Management Console, AWS CLI atau Amazon EMR API. Dalam tutorial ini, Anda menggunakan AWS CLI untuk bekerja dengan Iceberg di cluster EMR Amazon. Untuk menggunakan konsol untuk membuat cluster dengan Iceberg diinstal, ikuti langkah-langkah dalam Membangun 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 atau. Sediakan konfigurasi menggunakan AWS CLI saat Anda membuat sebuah klaster Sediakan konfigurasi menggunakan Java SDK ketika Anda membuat sebuah klaster

Atau, Anda dapat membuat cluster EMR Amazon 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 Mengirimkan 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 Ketergantungan dalam dokumen Apache Spark Running Spark di Kubernetes.

Inisialisasi sesi Spark untuk Iceberg

Contoh berikut menunjukkan cara meluncurkan shell Spark interaktif, menggunakan Spark submit, atau menggunakan Amazon EMR Notebooks 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 PySpark shell, 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 notebook EMR Studio, konfigurasikan sesi Spark Anda menggunakan perintah ajaib %%configure di notebook Amazon EMR Anda, seperti pada contoh berikut. Untuk informasi selengkapnya, lihat Menggunakan sihir EMR Notebooks 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 sampel berikut. Untuk informasi selengkapnya tentang menentukan klasifikasi konfigurasi menggunakan API EMR Amazon AWS CLI dan Amazon, lihat Mengonfigurasi 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 Iceberg

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

PySpark

Spark menyertakan shell berbasis Python,pyspark, yang dapat Anda gunakan untuk membuat prototipe program Spark yang ditulis dengan Python. Memanggil pyspark 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 tabel 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

Menggunakan Katalog Data AWS Glue dengan Spark Iceberg

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

Connect ke katalog AWS Glue default di wilayah default Anda

Contoh ini menunjukkan cara menghubungkan, menggunakan jenis katalog Glue. Jika Anda tidak menentukan ID katalog, ID tersebut 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

Connect ke katalog AWS Glue 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 Iceberg REST Catalog (IRC) dengan Spark Iceberg

Bagian yang mengikuti detail cara mengkonfigurasi integrasi Iceberg dengan katalog.

Hubungkan ke titik AWS akhir IRC Katalog Data Glue

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 klaster yang diaktifkan runtime-role, diperlukan pengaturan konfigurasi percikan tambahan berikut:

"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 AWS Glue endpoint dan kuota.

Connect ke endpoint IRC yang sewenang-wenang

Berikut ini menunjukkan contoh spark-submit perintah untuk menggunakan endpoint 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 menggunakan SparkCatalogsebagai 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:

Menggunakan Iceberg SparkSessionCatalog

Berikut ini menunjukkan perintah untuk menggunakan SparkSessionCatalogsebagai 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:

Menggunakan ekstensi Iceberg Spark

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

Pertimbangan untuk menggunakan Iceberg dengan Spark