使用DLF

步骤一：准备测试文件

创建一个简单的DataFrame，并将该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()

步骤二：上传测试文件

1. 进入资源上传页面。

   1. 登录E-MapReduce控制台。

   2. 在左侧导航栏，选择EMR Serverless > Spark。

   3. 在Spark页面，单击目标工作空间名称。

   4. 在EMR Serverless Spark页面，单击左侧导航栏中的文件管理。

2. 在文件管理页面，单击上传文件。

3. 在上传文件对话框中，单击待上传文件区域选择Python文件或者JAR包，或直接拖拽Python或文件者JAR包到待上传文件区域。

步骤三：创建批任务并运行

1. 在EMR Serverless Spark页面，单击左侧导航栏中的数据开发。

2. 在开发目录页签下，单击（新建）图标。

3. 在弹出的对话框中，输入名称，根据实际需求在批任务中选择PySpark或者JAR类型，然后单击确定。

4. 在右上角选择目标队列。

   添加队列的具体操作，请参见管理资源队列。

5. 在新建的开发页签中，配置以下信息，其余参数无需配置，然后单击运行。

   JAR

   参数	说明
   主jar资源	选择上传步骤一中编译的JAR包。
   Main Class	提交Spark任务时所指定的主类。本文示例填写为org.example.DlfAccess。

   PySpark

   参数	说明
   主Python资源	选择工作空间资源，然后选择在资源上传页面上传的Python文件。

步骤四：查看结果

任务运行完成后，您可以选择以下方式查看结果。