Uso de un clúster de Iceberg con Spark - Amazon EMR
Crear un clúster de IcebergInicio de una sesión de Spark para IcebergEscribir en una tabla de IcebergLeer desde una tabla de IcebergUso del catálogo de datos de AWS Glue con Spark IcebergUso del catálogo de REST de Iceberg (IRC) con Spark IcebergDiferencias de configuración cuando usas Iceberg SparkCatalog y SparkSessionCatalogUso de extensiones de Iceberg SparkConsideraciones para el uso de Iceberg con Spark

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Uso de un clúster de Iceberg con Spark

A partir de la versión 6.5.0 de Amazon EMR, puede usar Iceberg con su clúster de Spark sin necesidad de incluir acciones de arranque. Para las versiones 6.4.0 y anteriores de Amazon EMR, puede utilizar una acción de arranque para preinstalar todas las dependencias necesarias.

En este tutorial, utilizará el AWS CLI para trabajar con Iceberg en un clúster de Amazon EMR Spark. Para usar la consola para crear un clúster con Iceberg instalado, siga los pasos que se indican en Crear un lago de datos de Apache Iceberg con Amazon Athena, Amazon EMR y AWS Glue.

Crear un clúster de Iceberg

Puede crear un clúster con Iceberg instalado mediante la Consola de administración de AWS, la AWS CLI o la API de Amazon EMR. En este tutorial, utilizará el AWS CLI para trabajar con Iceberg en un clúster de Amazon EMR. Para usar la consola para crear un clúster con Iceberg instalado, siga los pasos que se indican en Crear un lago de datos de Apache Iceberg con Amazon Athena, Amazon EMR y AWS Glue.

Para usar Iceberg en Amazon EMR con AWS CLI el, cree primero un clúster con los siguientes pasos. Para obtener información sobre cómo especificar la clasificación de Iceberg mediante la AWS CLI, consulte o. Proporcione una configuración mediante la opción AWS CLI al crear un clúster Proporcione una configuración mediante el SDK de Java al crear un clúster

  1. Cree un archivo configurations.json, con el siguiente contenido:

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

  2. A continuación, cree un clúster con la siguiente configuración. Sustituya la ruta del bucket de Amazon S3 de ejemplo y el ID de subred por los suyos propios.

    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

Como alternativa, puede crear un clúster de Amazon EMR que incluya la aplicación de Spark e incluir el archivo /usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar como una dependencia JAR en un trabajo de Spark. Para obtener más información, consulte Envío de aplicaciones.

Para incluir el jar como una dependencia en un trabajo de Spark, agregue la siguiente propiedad de configuración a la aplicación de Spark:

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

Para obtener más información sobre las dependencias de los trabajos de Spark, consulte Administración de dependencias en el documento de Apache Spark Ejecución de Spark en Kubernetes.

Inicio de una sesión de Spark para Iceberg

En los siguientes ejemplos, se muestra cómo iniciar el intérprete de comandos interactivo de Spark, utilizar el envío de Spark o usar Cuadernos de Amazon EMR para trabajar con Amazon EMR.

spark-shell

  1. Conéctese al nodo principal utilizando SSH. Para obtener más información, consulte Conexión al nodo maestro mediante SSH en la Guía de administración de Amazon EMR.

  2. Introduzca el siguiente comando para iniciar el shell de Spark. Para usar la PySpark carcasa, sustitúyala porspark-shell. pyspark

    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. Conéctese al nodo principal utilizando SSH. Para obtener más información, consulte Conexión al nodo maestro mediante SSH en la Guía de administración de Amazon EMR.

  2. Ingrese el siguiente comando para iniciar la sesión de Spark para 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

Para iniciar una sesión de Spark con los Cuadernos de Amazon EMR, configure su sesión de Spark con el comando mágico %%configure de su cuaderno de Amazon EMR, como en el siguiente ejemplo. Para obtener más información, consulte Usar la magia de Cuadernos de Amazon EMR en la Guía de administración de 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

Ejecute el siguiente ejemplo para inicializar un clúster de Spark mediante la CLI y establecer todas las configuraciones predeterminadas de la sesión de Spark Iceberg. Para obtener más información sobre cómo especificar una clasificación de configuración mediante la API Amazon EMR AWS CLI y Amazon EMR, consulte Configurar aplicaciones.

[
  {
    "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"
    }
  }
]

Escribir en una tabla de Iceberg

En el siguiente ejemplo, se muestra cómo crear un conjunto de datos Iceberg DataFrame y cómo escribirlo como un conjunto de datos. En los ejemplos, se muestra cómo trabajar con conjuntos de datos mediante el intérprete de comandos de Spark mientras está conectado al nodo maestro usando SSH como usuario predeterminado de Hadoop.

nota

Para pegar muestras de código en el intérprete de comandos de Spark, escriba :paste en el símbolo del sistema, pegue el ejemplo y, a continuación, pulse CTRL+D.

PySpark

Spark incluye un intérprete de comandos basado en Python, pyspark, que puede utilizar para crear prototipos de programas de Spark escritos en Python. Invocar pyspark en el nodo maestro.

## 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()

Leer desde una tabla de Iceberg

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

Uso del catálogo de datos de AWS Glue con Spark Iceberg

Puedes conectarte al catálogo de datos de AWS Glue desde Spark Iceberg. En esta sección se muestran diferentes comandos para la conexión.

Conéctate al catálogo AWS Glue predeterminado de tu región predeterminada

En este ejemplo se muestra cómo realizar la conexión mediante el tipo de catálogo de Glue. Si no especifica un ID de catálogo, se utiliza el valor predeterminado:

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

Conectarse a un catálogo de AWS Glue con un identificador de catálogo específico

En este ejemplo se muestra cómo conectarse mediante un ID de catálogo:

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

Este comando se puede utilizar para conectarse a un catálogo de AWS Glue de una cuenta diferente, a un catálogo de RMS o a un catálogo federado.

Uso del catálogo de REST de Iceberg (IRC) con Spark Iceberg

En las secciones siguientes se detalla cómo configurar la integración de Iceberg con un catálogo.

Conéctese al AWS terminal IRC de Glue Data Catalog

A continuación se muestra un comando spark-submit de ejemplo para utilizar 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

Para usarlo en un clúster con funciones de tiempo de ejecución habilitadas, se necesitan los siguientes ajustes de configuración adicionales de Spark:

"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"

Para ver la lista de URL de terminales de AWS Glue para cada región, consulta Cuotas y puntos de conexión de AWS Glue.

Conéctese a un punto de conexión IRC arbitrario

A continuación se muestra un comando spark-submit de ejemplo para usar un punto de conexión de 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

Diferencias de configuración cuando usas Iceberg SparkCatalog y SparkSessionCatalog

Iceberg ofrece dos formas de crear catálogos de Spark Iceberg. Puede ajustar la configuración de Spark en SparkCatalog o en SparkSessionCatalog.

Uso de Iceberg SparkCatalog

A continuación se muestra el comando que se puede utilizar SparkCatalogcomo catálogo de 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

Consideraciones para este enfoque:

  • Puede acceder a las tablas de Iceberg, pero no a otras.

  • El nombre del catálogo no puede ser spark_catalog. Este es el nombre del catálogo inicial de Spark. Siempre se conecta a un metaalmacén de Hive. Es el catálogo predeterminado en Spark, a menos que el usuario lo sobrescriba mediante spark.sql.defaultCatalog.

  • Puede establecer el spark.sql.defaultCatalog al nombre de su catálogo para convertirlo en el predeterminado.

Uso de Iceberg SparkSessionCatalog

A continuación se muestra el comando que se puede utilizar SparkSessionCatalogcomo catálogo de 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

Consideraciones para este enfoque:

Uso de extensiones de Iceberg Spark

Iceberg ofrece la extensión Spark org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions que los usuarios pueden ajustar a través de la configuración de extensiones de Spark spark.sql.extensions. Las extensiones permiten funciones clave de Iceberg, como ELIMINAR, ACTUALIZAR y FUSIONAR a nivel de fila, instrucciones y procedimientos del lenguaje de definición de datos de Spark específicos de Iceberg, como la compactación, la caducidad de las instantáneas, la ramificación y el etiquetado, etc. Para obtener más detalles, consulte lo siguiente.

Consideraciones para el uso de Iceberg con Spark

  • De forma predeterminada, Amazon EMR 6.5.0 no admite la ejecución de Iceberg en Amazon EMR en EKS. Hay disponible una imagen personalizada de Amazon EMR 6.5.0 para que pueda pasar --jars local:///usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar como parámetro spark-submit para crear tablas de Iceberg en Amazon EMR en EKS. Para obtener más información, consulte Enviar una carga de trabajo de Spark en Amazon EMR con una imagen personalizada en la Guía de desarrollo de Amazon EMR en EKS. También puede contactar con Soporte si necesita ayuda. A partir de Amazon EMR 6.6.0, Iceberg es compatible con Amazon EMR en EKS.

  • Cuando utilice AWS Glue como catálogo de Iceberg, asegúrese de que la base de datos en la que va a crear la tabla esté en AWS Glue. Si utilizas servicios como estos AWS Lake Formation y no puedes cargar el catálogo, asegúrate de tener el acceso adecuado al servicio para ejecutar el comando.

  • Si usa Iceberg SparkSessionCatalog, como se describe enDiferencias de configuración cuando usas Iceberg SparkCatalog y SparkSessionCatalog, debe seguir los pasos de configuración descritos en Configurar el catálogo de datos de AWS Glue como el metaalmacén de Apache Hive, además de configurar los ajustes del catálogo de datos de Spark Iceberg AWS Glue.