E-MapReduce:DLF の使用

ステップ 1: テストファイルの準備

単純な DataFrame を作成し、pyspark_test という名前の Paimon テーブルに書き込みます。次に、Spark SQL を使用してテーブルデータをクエリし、データが正しく書き込まれていることを確認します。

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.12</artifactId>
    <version>3.5.2</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.12</artifactId>
    <version>3.5.2</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-hive_2.12</artifactId>
    <version>3.5.2</version>
    <scope>provided</scope>
</dependency>

package org.example;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.Arrays;
import java.util.List;

public class DlfAccess {
    public static void main(String[] args) {
        // SparkSession インスタンスを作成します。
        SparkSession spark = SparkSession.builder()
                .appName("DLF Example")
                .enableHiveSupport()
                .getOrCreate();

        // DataFrame を構築します。
        List<Row> data = Arrays.asList(
                RowFactory.create(1, "Alice"),
                RowFactory.create(2, "Bob"),
                RowFactory.create(3, "Charlie")
        );

        StructType schema = DataTypes.createStructType(new StructField[] {
                DataTypes.createStructField("id",   DataTypes.IntegerType, false),
                DataTypes.createStructField("name", DataTypes.StringType,  false)
        });
        // DataFrame を作成し、テーブルに書き込みます。

        Dataset<Row> df = spark.createDataFrame(data, schema);

        spark.sql("drop table if exists pyspark_test");
        df.write().format("paimon").mode("overwrite").saveAsTable("pyspark_test");

        // クエリを実行して、最初の 10 レコードを取得します。
        Dataset<Row> tableDf = spark.sql("select * from pyspark_test limit 10");

        tableDf.show();

        spark.stop();
    }
}

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("DLF test") \
    .enableHiveSupport() \
    .getOrCreate()

# テストデータ
data = [
    (1, "Alice"),
    (2, "Bob"),
    (3, "Charlie")
]

spark.sql("drop table if exists pyspark_test")
# DataFrame を作成し、テーブルに書き込みます。
df = spark.createDataFrame(data, schema='id int, name string')
df.write.format('paimon').mode("overwrite").saveAsTable("pyspark_test")

# テーブルをクエリして検証します。
spark.sql("select * from pyspark_test").show()

ステップ 2: テストファイルのアップロード

1. リソースアップロードページに移動します。

   1. EMR コンソールにログインします。

   2. 左側のナビゲーションウィンドウで、[EMR Serverless] > [Spark] を選択します。

   3. [Spark] ページで、ターゲットワークスペースの名前をクリックします。

   4. 左側のナビゲーションウィンドウで、[ファイル管理] をクリックします。

2. [ファイル管理] ページで、[ファイルのアップロード] をクリックします。

3. [ファイルのアップロード] ダイアログボックスで、アップロードエリアをクリックして Python ファイルまたは JAR パッケージを選択するか、ファイルをアップロードエリアにドラッグします。

ステップ 3: バッチタスクの作成と実行

1. 左側のナビゲーションウィンドウで、[データ開発] をクリックします。

2. [開発フォルダー] タブで、 (新規) アイコンをクリックします。

3. ダイアログボックスで、[名前] を入力します。[バッチタスク] セクションで、PySpark または JAR を選択します。次に、[OK] をクリックします。

4. 右上隅で、ターゲットキューを選択します。

   キューの追加方法の詳細については、「リソースキューの管理」をご参照ください。

5. 新しい開発タブで、次のパラメーターを構成し、他のパラメーターはデフォルト設定のままにして、[実行] をクリックします。

   JAR

   パラメーター	説明
   メイン JAR リソース	アップロードした JAR パッケージを選択します。
   メインクラス	Spark タスクを送信するときに指定するメインクラス。この例では、org.example.DlfAccess と入力します。

   PySpark

   パラメーター	説明
   メイン Python リソース	[ワークスペースリソース] を選択し、[リソースアップロード] ページでアップロードした Python ファイルを選択します。

ステップ 4: 結果の表示

タスクが完了したら、次のいずれかの方法で結果を表示できます。