E-MapReduce:Spark SQL 開発

前提条件

開始する前に、次のものが用意されていることを確認してください。

• ワークスペース。詳細については、「ワークスペースの作成」をご参照ください。

• SQL セッションインスタンス。詳細については、「SQL セッションの管理」をご参照ください。

Spark SQL ジョブの作成

1. 開発ページに移動します。

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

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

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

   4. [EMR Serverless Spark] ページで、左側のナビゲーションペインにある [Development] をクリックします。

2. ジョブを作成します。

   1. [Development] タブで、 アイコンをクリックします。

   2. ダイアログボックスで [Name] を入力し、[Type] を [Spark SQL] に設定して、[OK] をクリックします。

   3. 右上隅で、データカタログ、データベース、実行中の SQL セッションインスタンスを選択します。新しい SQL セッションインスタンスを作成するには、ドロップダウンリストから [Connect to SQL Session] を選択します。詳細については、「SQL セッションの管理」をご参照ください。

   4. エディタに SQL 文を入力します。

      例 1：基本的な SQL 操作

      データベースを作成して切り替え、テーブルを作成し、行を挿入して、データをクエリします。

      create DATABASE test_sql;
      
      use test_sql;
      
      CREATE TABLE IF NOT EXISTS my_table (id INT, name STRING);
      
      INSERT INTO my_table VALUES(1, 'Alice'),(2, 'Bob');
      
      SELECT * FROM my_table WHERE id > 1;

      結果はエディタ下部の出力パネルに表示されます。

      #	id	name
      1	2	Bob

      例 2：CSV ベースの外部テーブル

      Object Storage Service (OSS) の CSV ファイルに基づく外部テーブルを作成し、分析クエリを実行します。 oss://<bucketname>/user/ を実際のバケットパスに置き換えてください。

      1. 外部テーブルを作成します。テーブル名は orders とし、次のフィールドを定義します。

         • order_id: 注文 ID です。

         • order_date: 注文タイムスタンプ。例: '2025-07-01 10:00:00'。

         • order_category：商品カテゴリ。例：'電子機器'、'アパレル'。

         • order_revenue: 注文金額。

         CREATE TABLE orders (
           order_id       STRING,   -- 注文 ID
           order_date     STRING,   -- 注文タイムスタンプ
           order_category STRING,   -- 製品カテゴリ
           order_revenue  DOUBLE    -- 注文金額
         )
         USING CSV
         OPTIONS (
           path   'oss://<bucketname>/user/',
           header 'true'
         );

      2. テストデータを挿入します。

         INSERT OVERWRITE TABLE orders VALUES
           ('o1',  '2025-07-01 10:00:00', 'Electronics', 5999.0),
           ('o2',  '2025-07-02 11:30:00', 'Apparel',     299.0),
           ('o3',  '2025-07-03 14:45:00', 'Electronics', 899.0),
           ('o4',  '2025-07-04 09:15:00', 'Home Goods',  99.0),
           ('o5',  '2025-07-05 16:20:00', 'Electronics', 1999.0),
           ('o6',  '2025-07-06 08:00:00', 'Apparel',     199.0),
           ('o7',  '2025-07-07 12:10:00', 'Electronics', 799.0),
           ('o8',  '2025-07-08 18:30:00', 'Home Goods',  59.0),
           ('o9',  '2025-07-09 20:00:00', 'Electronics', 399.0),
           ('o10', '2025-07-10 07:45:00', 'Apparel',     599.0),
           ('o11', '2025-07-11 09:00:00', 'Electronics', 1299.0),
           ('o12', '2025-07-12 13:20:00', 'Home Goods',  159.0),
           ('o13', '2025-07-13 17:15:00', 'Apparel',     499.0),
           ('o14', '2025-07-14 21:30:00', 'Electronics', 999.0),
           ('o15', '2025-07-15 06:10:00', 'Home Goods',  299.0);

      3. 分析クエリを実行します。次のクエリは、15 日間のカテゴリ別売上実績 (注文数、流通取引総額 (GMV)、平均注文額、最新注文日時) を返します。対象は、総収益が 1,000 を超えるカテゴリで、結果は GMV の降順、カテゴリ名の昇順で並べ替えられます。

         SELECT
             order_category,
             COUNT(order_id)    AS order_count,
             SUM(order_revenue) AS gmv,
             AVG(order_revenue) AS avg_order_amount,
             MAX(order_date)    AS latest_order_date
         FROM orders
         WHERE
             CAST(order_date AS TIMESTAMP) BETWEEN '2025-07-01' AND '2025-07-15'
         GROUP BY order_category
         HAVING SUM(order_revenue) > 1000
         ORDER BY gmv DESC, order_category ASC;

      例 3：Hive Parquet テーブルの作成 (DLF フォーマットテーブル)

      デフォルトでは、EMR Serverless Spark は Apache Paimon をテーブルフォーマットとして使用します。代わりに Hive Parquet テーブルを作成するには、ジョブの起動パラメータでデフォルト設定を上書きする必要があります。

      起動パラメータ

      Hive Parquet テーブルを作成する前に、Paimon カタログがテーブル作成をインターセプトしないように、次の Spark 設定パラメータを追加します。

      spark.sql.catalog.spark_catalog=org.apache.spark.sql.hive.HiveCatalog
      spark.sql.defaultFileFormat=parquet

      これらのパラメーターは、EMR Serverless Spark コンソールの [設定] セクションで、または CLI で --conf パラメーターを使用して追加できます。

      SQL メソッド

      Hive Parquet テーブルを作成するには、CREATE TABLE ... USING PARQUET 構文を使用してください。STORED AS PARQUET は使用しないでください。

      CREATE TABLE my_parquet_table (
        id INT,
        name STRING,
        created_at TIMESTAMP
      ) USING PARQUET;

      DataFrame API メソッド

      データを Hive Parquet テーブルとして書き込むには、df.write.format("parquet").saveAsTable(...) を使用します。

      df.write.format("parquet").saveAsTable("my_parquet_table")

      説明

      ワークスペースが DLF Catalog にバインドされている場合は、Spark セッションまたは SQL 文を使用してテーブルを作成し、メタデータカタログとの一貫性を確保することを推奨します。

   5. (任意) 右側の [Version Information] タブをクリックしてバージョンを比較します。エディタでは、バージョン間の SQL コードの差異がハイライト表示されます。

3. ジョブの実行と公開を行います。

   1. [Run] をクリックします。結果は [Execution Results] タブに表示されます。エラーが発生した場合は、[Execution Issues] タブを確認してください。右側の実行履歴パネルには、過去 3 日間のレコードが表示されます。

   2. ジョブが正しく実行されることを確認した後、右上隅の [Publish] をクリックします。

   3. [Publish] ダイアログボックスで、リリースノートを入力し、[OK] をクリックします。

Spark UI へのアクセス

Spark UI には、タスクの実行ステータス、リソース使用量、ログ情報が表示されます。これらは Spark ジョブの分析と最適化に役立ちます。

キーボードショートカット

次のステップ

ワークフローを作成して、ジョブを定期的に実行するようにスケジュールします。詳細については、「ワークフローの作成」をご参照ください。完全なスケジューリングの例については、「Spark SQL 開発のクイックスタート」をご参照ください。

よくある質問