LAS Spark 任务适用于定时执行 Spark 离线任务的场景，支持 Jar 包资源和 Python资源引用的方式。

1 使用限制 #

LAS ByteLake 表的属性不支持并发写入数据，否则会出现写入冲突情况。通过 Spark 任务在向 ByteLake 表写入数据时，您可在 DataLeap 任务调度设置里，开启最大并发控制按钮，并将其设置为 1，以此避免因实例并发冲突致使任务执行失败。此外，您还需确保在其他业务中不存在并发写入同一个 ByteLake 表的情况。

2 使用前提 #

• 已开通湖仓一体分析服务 LAS 服务。详见湖仓一体分析服务 LAS 开通。
• 已在 DataLeap 项目控制台中，绑定相应的 湖仓一体分析服务（LAS）引擎实例。详见创建项目。

3 新建任务 #

1. 登录 DataLeap租户控制台，并进入项目的 IDE 开发界面，进行新建任务操作。
2. 单击新建任务按钮：
   1. 依次选择数据开发 > LAS > LAS Spark 任务类型。
   2. 关联实例：默认选择项目已绑定的 default 实例。绑定详见创建项目。
3. 填写任务基本信息，输入任务名称信息，如：las_spark_task，并选择任务存放的目标文件夹目录。单击确定按钮，完成任务创建。

4 任务配置说明 #

新建任务完成后，在任务配置界面完成以下参数配置。

4.1 语言设置 #

语言类型支持 Java、Python。

4.2 引入资源 #

• 语言类型选择 Java 时，资源类型支持 Jar 资源包的形式，可以按以下方式选择资源：

  • 从资源库选取已有的 jar 资源
  • 新建资源，详见：资源库。

• 语言类型选择 Python 时：

  • 资源类型默认选择 Python 类型，且需选择资源所关联的 LAS Schema 库信息。
  • 在编辑器中输入 Python 语句，执行引擎只支持 Python3.7。示例脚本如下：

  from pyspark import SparkFiles
  from pyspark.sql import SparkSession
  from pyspark.sql import SQLContext
  
  job_name='pyspark_test_on_las'
  
  spark = SparkSession.builder.appName(job_name).getOrCreate()
  spark.sql("select 1").show(10)
  
  spark.stop()
  

  更多 LAS Spark 任务操作，详见：Spark JAR 开发。

4.3 参数配置 #

4.4 任务产出登记 #

任务产出数据登记，用于记录任务---数据血缘信息，并不会对代码逻辑造成影响。对于系统无法通过解析获取产出信息的任务，可手动登记其产出信息。
如果任务含有 LAS 表的写入操作，强烈建议填写。您填写的内容即为任务产出，支持填写多个。其他任务的依赖推荐会根据此处填写的 LAS 表信息进行推荐。 具体登记内容包括：

• 数据类型：选择 LAS 时，需下拉选择当前任务写入数据的 LAS 数据库和数据表信息。
• 其他：该任务不写数据到 LAS 表。

5 使用示例 #

以下示例将为您演示如何通过 LAS Spark 任务中 Python 语言方式，来直接访问 LAS 表中的数据。

5.1 数据准备 #

1. 新建 LAS SQL 作业，操作详见：LAS SQL。
2. 在代码编辑区，编辑并执行以下示例语句，创建 LAS 示例表，并将数据写入表中：

CREATE  TABLE IF NOT EXISTS test_schema.student_demo (
    id INT COMMENT 'id',
    name STRING COMMENT 'name',
    age INT COMMENT 'age'
    )
    PARTITIONED BY (date STRING COMMENT 'date partition')
    stored as bytelake;
    
INSERT INTO test_schema.student_demo PARTITION (date = '20230518') VALUES(1, 'TOM', 10);

5.2 配置 LAS Spark 任务 #

1. 新建 LAS Spark 任务，详见上方新建任务。
2. 进入任务配置界面，语言类型选择 Python。
3. 选择库表资源所需关联的 LAS Schema 库信息：test_schema。
   
4. 在代码编辑区域，编辑以下相关 Python 查询语句：

   说明

   LAS ByteLake 2.0 和 Managed Hive 表类型，推荐使用以下 lasfs 示例；对于 ByteLake1.0 内表，需继续使用 Tunnel 方式。详见 Spark Jar 作业开发。

   from pyspark import SparkConf
   from pyspark.context import SparkContext
   from pyspark.sql import SparkSession, SQLContext
   
   SparkContext._ensure_initialized()
   spark = SparkSession.builder \
       .appName("pyspark_test_read_hive_x_new_1") \
       .config("spark.sql.lasfs.enabled", "true") \
       .config("spark.hadoop.hive.exec.scratchdir", "lasfs:/warehouse/tmp/hive") \
       .config("spark.hadoop.tmp.dir", "lasfs:/warehouse/tmp/hadoop") \
       .config("spark.hadoop.fs.lasfs.impl", "com.volcengine.las.fs.LasFileSystem") \
       .config("spark.hadoop.fs.lasfs.endpoint", "100.96.4.84:80") \
       .config("spark.hadoop.fs.lasfs.service.region", "cn-beijing") \
       .config("spark.hadoop.fs.lasfs.access.key", "AKxxx") \
       .config("spark.hadoop.fs.lasfs.secret.key", "SKxxx==") \
       .config("spark.sql.analysis.hoodie.relation.conversion.skip.enabled", "false") \
       .config("spark.sql.extensions", "org.apache.spark.sql.LASExtension") \
       .enableHiveSupport() \
       .getOrCreate()
   sc = spark.sparkContext
   sql = spark.sql
   sqlContext = spark._wrapped
   sqlCtx = sqlContext
   
   df = sql("insert into test_schema.student_demo partition (date='20240619') values('1','TOM','10')")
   df.show()
   
   spark.stop()
   

• 替换参数说明：

  参数	说明
  appName	自定义 Spark 任务名称。
  access.key	可前往访问控制的密钥管理中查看 Access Key ID。相关操作说明请参见 Access Key（密钥）管理。
  secret.key	可前往访问控制的密钥管理中查看 Secret Access Key。相关操作说明请参见 Access Key（密钥）管理。
  endpoint	需按照您所在的 Region 信息进行配置，不同地域值，详见 Spark Jar 作业开发 - spark.hadoop.fs.lasfs.endpoint。
  region	按照您所在的 Region 信息，对应配置： 华北region：cn-beijing 华东region：cn-shanghai 华南region：cn-grougzhou 柔佛region：ap-southeast-1