翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。 # Spark で Iceberg クラスターを使用する Amazon EMR バージョン 6.5.0 以降では、ブートストラップアクションを追加しなくても Spark クラスターで Iceberg を使用できます。Amazon EMR バージョン 6.4.0 以前の場合、ブートストラップアクションを使用して必要なすべての依存関係を事前インストールできます。 このチュートリアルでは、 AWS CLI を使用して Amazon EMR Spark クラスターで Iceberg を操作します。コンソールを使用して Iceberg がインストールされたクラスターを作成するには、「[Build an Apache Iceberg data lake using Amazon Athena, Amazon EMR, and AWS Glue](https://aws.amazon.com/blogs//big-data/build-an-apache-iceberg-data-lake-using-amazon-athena-amazon-emr-and-aws-glue/)」の手順に従ってください。 ## Iceberg クラスターの作成 Iceberg がインストールされたクラスターは、、 AWS マネジメントコンソール、 AWS CLI または Amazon EMR API を使用して作成できます。このチュートリアルでは、 AWS CLI を使用して Amazon EMR クラスターで Iceberg を操作します。コンソールを使用して Iceberg がインストールされたクラスターを作成するには、「[Build an Apache Iceberg data lake using Amazon Athena, Amazon EMR, and AWS Glue](https://aws.amazon.com/blogs//big-data/build-an-apache-iceberg-data-lake-using-amazon-athena-amazon-emr-and-aws-glue/)」の手順に従ってください。 で Amazon EMR で Iceberg を使用するには AWS CLI、まず次のステップでクラスターを作成します。を使用して Iceberg 分類を指定する方法については AWS CLI、[クラスターの作成 AWS CLI 時に を使用して設定を指定する](emr-configure-apps-create-cluster.md#emr-configure-apps-create-cluster-cli)「」または「」を参照してください[クラスター作成時に Java SDK を使用して設定を指定する](emr-configure-apps-create-cluster.md#emr-configure-apps-create-cluster-sdk)。 1. 以下のコンテンツを含む `configurations.json` ファイルを作成します。 ``` [{ "Classification":"iceberg-defaults", "Properties":{"iceberg.enabled":"true"} }] ``` 1. 次に、以下の設定でクラスターを作成します。この例の Amazon S3 バケットパスとサブネット ID は、実際の値に置き換えてください。 ``` 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}} ``` または、Spark アプリケーションを含む Amazon EMR クラスターを作成し、`/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar` ファイルを Spark ジョブの JAR 依存関係として追加することもできます。詳細については、「[Submitting Applications](https://spark.apache.org/docs/latest/submitting-applications.html#submitting-applications)」を参照してください。 この jar を Spark ジョブの依存関係として含めるには、以下の設定プロパティを Spark アプリケーションに追加します。 ``` --conf "spark.jars=/usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar" ``` Spark ジョブの依存関係の詳細については、Apache Spark ドキュメント「[Running Spark on Kubernetes](https://spark.apache.org/docs/latest/running-on-kubernetes.html)」の「[Dependency Management](https://spark.apache.org/docs/latest/running-on-kubernetes.html#dependency-management)」を参照してください。 ## Iceberg の Spark セッションを初期化する 以下の例では、インタラクティブな Spark シェルを起動し、Spark submit を使用するか、Amazon EMR Notebooks を使用して、Amazon EMR で Iceberg を操作する方法を示します。 ------ #### [ spark-shell ] 1. SSH を使用してマスターノードに接続します。詳細については、「*Amazon EMR 管理ガイド*」の「[SSH を使用してマスターノードに接続する](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html)」を参照してください。 1. 以下のコマンドを入力して、Spark シェルを起動します。PySpark シェルを使用するには、`spark-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. SSH を使用してマスターノードに接続します。詳細については、「*Amazon EMR 管理ガイド*」の「[SSH を使用してマスターノードに接続する](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-connect-master-node-ssh.html)」を参照してください。 1. 以下のコマンドを入力して、Iceberg の Spark セッションを起動します。 ``` 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 ] EMR Studio ノートブックを使用して Spark セッションを初期化するには、次の例のように、Amazon EMR Notebooks で `%%configure` マジックコマンドを使用して Spark セッションを設定します。詳細については、「*Amazon EMR 管理ガイド*」の「[Use EMR Notebooks magics](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-studio-magics.html#emr-magics)」を参照してください。 ``` %%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 ] CLI を使用して Spark クラスターを初期化し、Spark Iceberg セッションのデフォルト設定をすべて設定するには、以下のサンプルを実行します。 AWS CLI と Amazon EMR API を使用して設定分類を指定する方法の詳細については、[「アプリケーションの設定](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html)」を参照してください。 ``` [ { "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" } } ] ``` ------ ## Iceberg テーブルへの書き込み 以下の例では、DataFrame を作成し、それを Iceberg データセットとして書き込む方法を示します。この例では、デフォルトの Hadoop ユーザーとして SSH を使用してマスターノードに接続しながら、Spark シェルを使用してデータセットを操作する方法を示しています。 **注記** コードサンプルを Spark シェルに貼り付けるには、プロンプトで「`:paste`」と入力し、例を貼り付けて、[`CTRL+D`] を押します。 ------ #### [ PySpark ] Spark には、Python ベースのシェルである `pyspark` 用意されており、Python で記述された Spark プログラムのプロトタイプを作成するために使用できます。マスターノードで `pyspark` を起動します。 ``` ## 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() ``` ------ ## 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 ``` ------ ## Spark Iceberg AWS での Glue データカタログの使用 Spark Iceberg から AWS Glue データカタログに接続できます。このセクションでは、接続するためのさまざまなコマンドを示します。 ### デフォルトのリージョンのデフォルトの AWS Glue カタログに接続する このサンプルは、 Glue カタログタイプを使用して接続する方法を示しています。カタログ ID を指定しない場合、デフォルトが使用されます: ``` 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 ``` ### 特定のカタログ ID AWS を使用して Glue カタログに接続する このサンプルは、カタログ ID を使用して接続する方法を示しています: ``` 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 ``` このコマンドは、別のアカウントの AWS Glue カタログ、RMS カタログ、またはフェデレーティッドカタログへの接続に使用できます。 ## Spark Iceberg での Iceberg REST Catalog (IRC) の使用 以下のセクションでは、Iceberg とカタログの統合を設定する方法について詳しく説明します。 ### Glue データカタログ IRC AWS エンドポイントに接続する Iceberg REST を使用するための `spark-submit` コマンドの例を以下に示します: ``` 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 ``` ランタイムロールが有効なクラスターで使用するには、次の追加の 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" ``` 各リージョンの AWS Glue エンドポイント URL リストについては、「 [AWS Glue エンドポイントとクォータ](https://docs.aws.amazon.com/general/latest/gr/glue.html)」を参照してください。 ### 任意の IRC エンドポイントに接続する IRC エンドポイントを使用するための `spark-submit` コマンドの例を以下に示します: ``` 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 ``` ## Iceberg SparkCatalog と SparkSessionCatalog を使用する場合の設定の違い Iceberg では、Spark Iceberg カタログを作成する 2 つの方法を利用できます。Spark 設定は、`SparkCatalog` または `SparkSessionCatalog` のいずれかに設定できます。 ### Iceberg SparkCatalog の使用 Spark Iceberg カタログとして **SparkCatalog** を使用するためのコマンドを以下に示します: ``` 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 ``` このアプローチに関する考慮事項: + Iceberg テーブルにはアクセスできますが、他のテーブルにはアクセスできません。 + カタログ名を「**spark\_catalog**」にすることはできません。これは Spark の初期カタログの名前です。常に Hive メタストアに接続します。これは、ユーザーが `spark.sql.defaultCatalog` を使用して上書きしない限り、Spark のデフォルトカタログです。 + `spark.sql.defaultCatalog` をカタログ名に設定して、デフォルトのカタログにができます。 ### Iceberg SparkSessionCatalog の使用 Spark Iceberg カタログとして **[SparkSessionCatalog]** を使用するためのコマンドを以下に示します。 ``` 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 ``` このアプローチに関する考慮事項: + テーブルが Iceberg テーブルとして見つからない場合、Spark は Hive メタストア内のテーブルであるかどうかを確認しようとします。詳細については、[「Hive AWS のカタログとしての Glue データカタログの使用](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-hive-metastore-glue.html)」を参照してください。 + カタログ名は「**spark\_catalog**」である必要があります。 ## Iceberg Spark 拡張機能の使用 Iceberg は、ユーザーが Spark 拡張機能設定 `spark.sql.extensions` を通じて設定できる Spark 拡張機能 `org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions` を提供します。拡張機能により、行レベルの DELETE、UPDATE、MERGE などの Iceberg の主要な機能、圧縮、スナップショットの有効期限、分岐とタグ付けなどの Iceberg 固有の Spark データ定義言語ステートメントと手順が有効になります。詳細については、以下をご参照ください。 + Iceberg Spark 書き込み拡張機能:「[Spark Writes](https://iceberg.apache.org/docs/nightly/spark-writes/)」 + Iceberg Spark DDL 拡張機能:「[ALTER TABLE SQL extensions](https://iceberg.apache.org/docs/nightly/spark-ddl/#alter-table-sql-extensions/)」 + Iceberg Spark プロシージャ拡張機能:「[Spark Procedures](https://iceberg.apache.org/docs/nightly/spark-ddl/#alter-table-sql-extensions/)」 ## Spark で Iceberg を使用するための考慮事項 + Amazon EMR 6.5.0 は、デフォルトでは Amazon EMR on EKS での Iceberg の実行をサポートしていません。Amazon EMR 6.5.0 カスタムイメージが用意されているため、`--jars local:///usr/share/aws/iceberg/lib/iceberg-spark3-runtime.jar` を `spark-submit` パラメータとして渡して Amazon EMR on EKS に Iceberg テーブルを作成できます。詳細については、「*Amazon EMR on EKS 開発ガイド*」の「[Submit a Spark workload in Amazon EMR using a custom image](https://docs.aws.amazon.com/emr/latest/EMR-on-EKS-DevelopmentGuide/docker-custom-images-steps.html#docker-custom-images-submit)」を参照してください。 サポート に問い合わせることもできます。Amazon EMR 6.6.0 以降、Iceberg は Amazon EMR on EKS でサポートされます。 + Iceberg のカタログとして AWS Glue を使用する場合は、テーブルを作成するデータベースが Glue AWS に存在することを確認してください。などのサービスを使用して AWS Lake Formation いて、カタログをロードできない場合は、コマンドを実行するためのサービスへの適切なアクセス権があることを確認してください。 + 「」で説明されているように Iceberg SparkSessionCatalog を使用する場合は[Iceberg SparkCatalog と SparkSessionCatalog を使用する場合の設定の違い](#emr-iceberg-spark-catalog)、Spark Iceberg [AWS Glue データカタログの設定に加えて、「Apache Hive メタストアとして](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-hive-metastore-glue.html) AWS Glue データカタログを設定する」で説明されている設定ステップに従う必要があります。