Menggunakan kerangka Hudi di Glue AWS - AWS Glue
Mengaktifkan HudiContoh: Tulis HudiContoh: Baca HudiContoh: Perbarui dan masukkan ke dalam tabel HudiContoh: Membaca tabel Hudi menggunakan SparkContoh: Tulis tabel Hudi menggunakan SparkContoh: Membaca dan menulis tabel Hudi dengan kontrol izin Lake Formation

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

Menggunakan kerangka Hudi di Glue AWS

AWSGlue 3.0 dan yang lebih baru mendukung kerangka Apache Hudi untuk data lake. Hudi adalah kerangka penyimpanan danau data open-source yang menyederhanakan pemrosesan data tambahan dan pengembangan pipa data. Topik ini mencakup fitur yang tersedia untuk menggunakan data Anda di AWS Glue saat Anda mengangkut atau menyimpan data Anda dalam tabel Hudi. Untuk mempelajari lebih lanjut tentang Hudi, lihat dokumentasi resmi Apache Hudi.

Anda dapat menggunakan AWS Glue untuk melakukan operasi baca dan tulis pada tabel Hudi di Amazon S3, atau bekerja dengan tabel Hudi menggunakan Katalog Data AWS Glue. Operasi tambahan termasuk insert, update, dan semua operasi Apache Spark juga didukung.

catatan

Apache Hudi 0.10.1 untuk AWS Glue 3.0 tidak mendukung tabel Hudi Merge on Read (MoR).

Tabel berikut mencantumkan versi Hudi yang disertakan dalam setiap versi AWS Glue.

AWSVersi Glue Versi Hudi yang didukung
4.0 0.12.1
3.0 0.10.1

Untuk mempelajari lebih lanjut tentang framework data lake yang didukung AWS Glue, lihatMenggunakan kerangka kerja data lake dengan pekerjaan AWS Glue ETL.

Mengaktifkan Hudi

Untuk mengaktifkan Hudi for AWS Glue, selesaikan tugas berikut:

  • Tentukan hudi sebagai nilai untuk parameter --datalake-formats pekerjaan. Untuk informasi selengkapnya, lihat AWS Glueparameter pekerjaan.

  • Buat kunci bernama --conf untuk pekerjaan AWS Glue Anda, dan atur ke nilai berikut. Atau, Anda dapat mengatur konfigurasi berikut menggunakan SparkConf skrip Anda. Pengaturan ini membantu Apache Spark menangani tabel Hudi dengan benar.

    spark.serializer=org.apache.spark.serializer.KryoSerializer --conf spark.sql.hive.convertMetastoreParquet=false

  • Dukungan izin Lake Formation untuk Hudi diaktifkan secara default untuk AWS Glue 4.0. Tidak diperlukan konfigurasi tambahan untuk membaca/menulis ke tabel Hudi yang terdaftar di Lake Formation. Untuk membaca tabel Hudi yang terdaftar, peran IAM AWS Glue job harus memiliki izin SELECT. Untuk menulis ke tabel Hudi terdaftar, peran IAM pekerjaan AWS Glue harus memiliki izin SUPER. Untuk mempelajari lebih lanjut tentang mengelola izin Lake Formation, lihat Memberikan dan mencabut izin pada sumber daya Katalog Data.

Menggunakan versi Hudi yang berbeda

Untuk menggunakan versi Hudi yang tidak didukung AWS Glue, tentukan file Hudi JAR Anda sendiri menggunakan parameter --extra-jars pekerjaan. Jangan sertakan hudi sebagai nilai untuk parameter --datalake-formats pekerjaan.

Contoh: Tulis tabel Hudi ke Amazon S3 dan daftarkan di Katalog Data AWS Glue

Contoh skrip ini menunjukkan cara menulis tabel Hudi ke Amazon S3 dan mendaftarkan tabel ke Katalog Data AWS Glue. Contoh menggunakan alat Hudi Hive Sync untuk mendaftarkan tabel.

catatan

Contoh ini mengharuskan Anda untuk mengatur parameter --enable-glue-datacatalog pekerjaan untuk menggunakan Katalog Data AWS Glue sebagai metastore Apache Spark Hive. Untuk mempelajari selengkapnya, lihat AWS Glueparameter pekerjaan.

Python
# Example: Create a Hudi table from a DataFrame 
# and register the table to Glue Data Catalog

additional_options={
    "hoodie.table.name": "<your_table_name>",
    "hoodie.datasource.write.storage.type": "COPY_ON_WRITE",
    "hoodie.datasource.write.operation": "upsert",
    "hoodie.datasource.write.recordkey.field": "<your_recordkey_field>",
    "hoodie.datasource.write.precombine.field": "<your_precombine_field>",
    "hoodie.datasource.write.partitionpath.field": "<your_partitionkey_field>",
    "hoodie.datasource.write.hive_style_partitioning": "true",
    "hoodie.datasource.hive_sync.enable": "true",
    "hoodie.datasource.hive_sync.database": "<your_database_name>",
    "hoodie.datasource.hive_sync.table": "<your_table_name>",
    "hoodie.datasource.hive_sync.partition_fields": "<your_partitionkey_field>",
    "hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor",
    "hoodie.datasource.hive_sync.use_jdbc": "false",
    "hoodie.datasource.hive_sync.mode": "hms",
    "path": "s3://<s3Path/>"
}

dataFrame.write.format("hudi") \
    .options(**additional_options) \
    .mode("overwrite") \
    .save()
Scala
// Example: Example: Create a Hudi table from a DataFrame
// and register the table to Glue Data Catalog

val additionalOptions = Map(
  "hoodie.table.name" -> "<your_table_name>",
  "hoodie.datasource.write.storage.type" -> "COPY_ON_WRITE",
  "hoodie.datasource.write.operation" -> "upsert",
  "hoodie.datasource.write.recordkey.field" -> "<your_recordkey_field>",
  "hoodie.datasource.write.precombine.field" -> "<your_precombine_field>",
  "hoodie.datasource.write.partitionpath.field" -> "<your_partitionkey_field>",
  "hoodie.datasource.write.hive_style_partitioning" -> "true",
  "hoodie.datasource.hive_sync.enable" -> "true",
  "hoodie.datasource.hive_sync.database" -> "<your_database_name>",
  "hoodie.datasource.hive_sync.table" -> "<your_table_name>",
  "hoodie.datasource.hive_sync.partition_fields" -> "<your_partitionkey_field>",
  "hoodie.datasource.hive_sync.partition_extractor_class" -> "org.apache.hudi.hive.MultiPartKeysValueExtractor",
  "hoodie.datasource.hive_sync.use_jdbc" -> "false",
  "hoodie.datasource.hive_sync.mode" -> "hms",
  "path" -> "s3://<s3Path/>")

dataFrame.write.format("hudi")
  .options(additionalOptions)
  .mode("append")
  .save()

Contoh: Membaca tabel Hudi dari Amazon S3 menggunakan Katalog Data AWS Glue

Contoh ini membaca tabel Hudi yang Anda buat Contoh: Tulis tabel Hudi ke Amazon S3 dan daftarkan di Katalog Data AWS Glue dari Amazon S3.

catatan

Contoh ini mengharuskan Anda untuk mengatur parameter --enable-glue-datacatalog pekerjaan untuk menggunakan Katalog Data AWS Glue sebagai metastore Apache Spark Hive. Untuk mempelajari selengkapnya, lihat AWS Glueparameter pekerjaan.

Python

Untuk contoh ini, gunakan GlueContext.create_data_frame.from_catalog() metode ini.

# Example: Read a Hudi table from Glue Data Catalog

from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)

dataFrame = glueContext.create_data_frame.from_catalog(
    database = "<your_database_name>",
    table_name = "<your_table_name>"
)
Scala

Untuk contoh ini, gunakan getCatalogSourcemetode ini.

// Example: Read a Hudi table from Glue Data Catalog

import com.amazonaws.services.glue.GlueContext
import org.apache.spark.SparkContext

object GlueApp {
  def main(sysArgs: Array[String]): Unit = {
    val spark: SparkContext = new SparkContext()
    val glueContext: GlueContext = new GlueContext(spark)
    
    val dataFrame = glueContext.getCatalogSource(
      database = "<your_database_name>",
      tableName = "<your_table_name>"
    ).getDataFrame()
  }
}

Contoh: Perbarui dan masukkan DataFrame ke dalam tabel Hudi di Amazon S3

Contoh ini menggunakan AWS Glue Data Catalog untuk menyisipkan DataFrame ke dalam tabel Hudi yang Anda buat. Contoh: Tulis tabel Hudi ke Amazon S3 dan daftarkan di Katalog Data AWS Glue

catatan

Contoh ini mengharuskan Anda untuk mengatur parameter --enable-glue-datacatalog pekerjaan untuk menggunakan Katalog Data AWS Glue sebagai metastore Apache Spark Hive. Untuk mempelajari selengkapnya, lihat AWS Glueparameter pekerjaan.

Python

Untuk contoh ini, gunakan GlueContext.write_data_frame.from_catalog() metode ini.

# Example: Upsert a Hudi table from Glue Data Catalog

from awsglue.context import GlueContext
from pyspark.context import SparkContext

sc = SparkContext()
glueContext = GlueContext(sc)

glueContext.write_data_frame.from_catalog(
    frame = dataFrame,
    database = "<your_database_name>",
    table_name = "<your_table_name>",
    additional_options={
        "hoodie.table.name": "<your_table_name>",
        "hoodie.datasource.write.storage.type": "COPY_ON_WRITE",
        "hoodie.datasource.write.operation": "upsert",
        "hoodie.datasource.write.recordkey.field": "<your_recordkey_field>",
        "hoodie.datasource.write.precombine.field": "<your_precombine_field>",
        "hoodie.datasource.write.partitionpath.field": "<your_partitionkey_field>",
        "hoodie.datasource.write.hive_style_partitioning": "true",
        "hoodie.datasource.hive_sync.enable": "true",
        "hoodie.datasource.hive_sync.database": "<your_database_name>",
        "hoodie.datasource.hive_sync.table": "<your_table_name>",
        "hoodie.datasource.hive_sync.partition_fields": "<your_partitionkey_field>",
        "hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor",
        "hoodie.datasource.hive_sync.use_jdbc": "false",
        "hoodie.datasource.hive_sync.mode": "hms"
    }
)
Scala

Untuk contoh ini, gunakan getCatalogSinkmetode ini.

// Example: Upsert a Hudi table from Glue Data Catalog

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.JsonOptions
import org.apacke.spark.SparkContext

object GlueApp {
  def main(sysArgs: Array[String]): Unit = {
    val spark: SparkContext = new SparkContext()
    val glueContext: GlueContext = new GlueContext(spark)
    glueContext.getCatalogSink("<your_database_name>", "<your_table_name>",
      additionalOptions = JsonOptions(Map(
        "hoodie.table.name" -> "<your_table_name>",
        "hoodie.datasource.write.storage.type" -> "COPY_ON_WRITE",
        "hoodie.datasource.write.operation" -> "upsert",
        "hoodie.datasource.write.recordkey.field" -> "<your_recordkey_field>",
        "hoodie.datasource.write.precombine.field" -> "<your_precombine_field>",
        "hoodie.datasource.write.partitionpath.field" -> "<your_partitionkey_field>",
        "hoodie.datasource.write.hive_style_partitioning" -> "true",
        "hoodie.datasource.hive_sync.enable" -> "true",
        "hoodie.datasource.hive_sync.database" -> "<your_database_name>",
        "hoodie.datasource.hive_sync.table" -> "<your_table_name>",
        "hoodie.datasource.hive_sync.partition_fields" -> "<your_partitionkey_field>",
        "hoodie.datasource.hive_sync.partition_extractor_class" -> "org.apache.hudi.hive.MultiPartKeysValueExtractor",
        "hoodie.datasource.hive_sync.use_jdbc" -> "false",
        "hoodie.datasource.hive_sync.mode" -> "hms"
      )))
      .writeDataFrame(dataFrame, glueContext)
  }
}

Contoh: Baca Tabel Hudi dari Amazon S3 menggunakan Spark

Contoh ini membaca tabel Hudi dari Amazon S3 menggunakan Spark API. DataFrame

Python
# Example: Read a Hudi table from S3 using a Spark DataFrame

dataFrame = spark.read.format("hudi").load("s3://<s3path/>")
Scala
// Example: Read a Hudi table from S3 using a Spark DataFrame

val dataFrame = spark.read.format("hudi").load("s3://<s3path/>")

Contoh: Tabel Hudi ke Amazon S3 menggunakan Spark

Contoh ini menulis tabel Hudi ke Amazon S3 menggunakan Spark.

Python
# Example: Write a Hudi table to S3 using a Spark DataFrame

dataFrame.write.format("hudi") \
    .options(**additional_options) \
    .mode("overwrite") \
    .save("s3://<s3Path/>)
Scala
// Example: Write a Hudi table to S3 using a Spark DataFrame

dataFrame.write.format("hudi")
  .options(additionalOptions)
  .mode("overwrite")
  .save("s3://<s3path/>")

Contoh: Membaca dan menulis tabel Hudi dengan kontrol izin Lake Formation

Contoh ini membaca dan menulis tabel Hudi dengan kontrol izin Lake Formation.

  1. Buat tabel Hudi dan daftarkan di Lake Formation.

    1. Untuk mengaktifkan kontrol izin Lake Formation, Anda harus terlebih dahulu mendaftarkan tabel jalur Amazon S3 di Lake Formation. Untuk informasi selengkapnya, lihat Mendaftarkan lokasi Amazon S3. Anda dapat mendaftarkannya baik dari konsol Lake Formation atau dengan menggunakan AWS CLI:

      aws lakeformation register-resource --resource-arn arn:aws:s3:::<s3-bucket>/<s3-folder> --use-service-linked-role --region <REGION>

      Setelah Anda mendaftarkan lokasi Amazon S3, tabel AWS Glue apa pun yang menunjuk ke lokasi (atau lokasi turunannya) akan mengembalikan nilai IsRegisteredWithLakeFormation parameter sebagai true dalam panggilan. GetTable

    2. Buat tabel Hudi yang menunjuk ke jalur Amazon S3 terdaftar melalui API kerangka data Spark:

      hudi_options = {
    'hoodie.table.name': table_name,
    'hoodie.datasource.write.storage.type': 'COPY_ON_WRITE',
    'hoodie.datasource.write.recordkey.field': 'product_id',
    'hoodie.datasource.write.table.name': table_name,
    'hoodie.datasource.write.operation': 'upsert',
    'hoodie.datasource.write.precombine.field': 'updated_at',
    'hoodie.datasource.write.hive_style_partitioning': 'true',
    'hoodie.upsert.shuffle.parallelism': 2,
    'hoodie.insert.shuffle.parallelism': 2,
    'path': <S3_TABLE_LOCATION>,
    'hoodie.datasource.hive_sync.enable': 'true',
    'hoodie.datasource.hive_sync.database': database_name,
    'hoodie.datasource.hive_sync.table': table_name,
    'hoodie.datasource.hive_sync.use_jdbc': 'false',
    'hoodie.datasource.hive_sync.mode': 'hms'
}

df_products.write.format("hudi")  \
    .options(**hudi_options)  \
    .mode("overwrite")  \
    .save()

  2. Berikan izin Formasi Lake untuk peran IAM pekerjaan AWS Glue. Anda dapat memberikan izin dari konsol Lake Formation, atau menggunakan AWS CLI. Untuk informasi selengkapnya, lihat Memberikan izin tabel menggunakan konsol Lake Formation dan metode sumber daya bernama

  3. Baca tabel Hudi yang terdaftar di Lake Formation. Kodenya sama dengan membaca tabel Hudi yang tidak terdaftar. Perhatikan bahwa peran IAM AWS Glue job harus memiliki izin SELECT agar pembacaan berhasil.

     val dataFrame = glueContext.getCatalogSource(
      database = "<your_database_name>",
      tableName = "<your_table_name>"
    ).getDataFrame()

  4. Tulis ke tabel Hudi yang terdaftar di Lake Formation. Kode ini sama dengan menulis ke tabel Hudi yang tidak terdaftar. Perhatikan bahwa peran IAM AWS Glue job harus memiliki izin SUPER agar penulisan berhasil.

    glueContext.getCatalogSink("<your_database_name>", "<your_table_name>",
      additionalOptions = JsonOptions(Map(
        "hoodie.table.name" -> "<your_table_name>",
        "hoodie.datasource.write.storage.type" -> "COPY_ON_WRITE",
        "hoodie.datasource.write.operation" -> "<write_operation>",
        "hoodie.datasource.write.recordkey.field" -> "<your_recordkey_field>",
        "hoodie.datasource.write.precombine.field" -> "<your_precombine_field>",
        "hoodie.datasource.write.partitionpath.field" -> "<your_partitionkey_field>",
        "hoodie.datasource.write.hive_style_partitioning" -> "true",
        "hoodie.datasource.hive_sync.enable" -> "true",
        "hoodie.datasource.hive_sync.database" -> "<your_database_name>",
        "hoodie.datasource.hive_sync.table" -> "<your_table_name>",
        "hoodie.datasource.hive_sync.partition_fields" -> "<your_partitionkey_field>",
        "hoodie.datasource.hive_sync.partition_extractor_class" -> "org.apache.hudi.hive.MultiPartKeysValueExtractor",
        "hoodie.datasource.hive_sync.use_jdbc" -> "false",
        "hoodie.datasource.hive_sync.mode" -> "hms"
      )))
      .writeDataFrame(dataFrame, glueContext)