开发ODPS Spark任务

使用限制

当ODPS Spark节点选择Spark3.x版本时，若提交节点报错，请购买使用Serverless资源组，详情请参见新增和使用Serverless资源组。

准备工作

ODPS Spark节点支持使用Java/Scala和Python语言运行Spark on MaxCompute离线作业，不同语言开发步骤及配置界面存在差异，您可根据业务需要选择使用。

Java/Scala

在ODPS Spark节点执行Java或Scala语言类型代码前，您需先在本地开发好Spark on MaxCompute作业代码，再通过DataWorks上传为MaxCompute的资源。步骤如下：

1. 准备开发环境。

   根据所使用系统类型，准备运行Spark on MaxCompute任务的开发环境，详情请参见搭建Linux开发环境、搭建Windows开发环境。

2. 开发Java/Scala代码。

   在ODPS Spark节点执行Java或Scala语言类型代码前，需先在本地或已有环境开发好Spark on MaxCompute代码，建议使用Spark on MaxCompute提供的项目示例工程模板进行开发。

3. 打包代码并上传至DataWorks。

   代码开发完成后，需将其打包，并通过DataWorks上传为MaxCompute资源，详情请参见创建并使用MaxCompute资源。

Python（使用默认Python环境实现）

Python（使用自定义Python环境实现）

若平台提供的默认Python环境无法满足您的业务需求，则可根据如下步骤自定义Python环境，执行Spark on MaxCompute任务。

1. 本地准备Python环境。

   您可参考PySpark Python版本和依赖支持，根据业务需要配置可用的Python环境。

2. 打包环境并上传至DataWorks。

   将Python环境压缩为一个Zip包，并通过DataWorks上传为MaxCompute资源，作为后续运行Spark on MaxCompute任务的执行环境。详情请参见创建并使用MaxCompute资源。

配置项说明

DataWorks运行Spark on MaxCompute离线作业采用Cluster模式，在Cluster模式中，您需指定自定义程序入口main。main运行结束（即状态为Success或Fail）时，对应的Spark作业便会结束。此外，spark-defaults.conf中的配置需逐条加到ODPS Spark节点配置项中。例如，Executor的数量、内存大小和spark.hadoop.odps.runtime.end.point的配置。

编辑代码：简单示例

以下以一个简单示例为您介绍ODPS Spark节点的使用：判断一个字符串是否可以转换为数字。

1. 创建资源。

   1. 在数据开发页面新建Python类型的资源，并命名为spark_is_number.py，详情请参见创建并使用MaxCompute资源。代码如下：

      # -*- coding: utf-8 -*-
      import sys
      from pyspark.sql import SparkSession
      
      try:
          # for python 2
          reload(sys)
          sys.setdefaultencoding('utf8')
      except:
          # python 3 not needed
          pass
      
      if __name__ == '__main__':
          spark = SparkSession.builder\
              .appName("spark sql")\
              .config("spark.sql.broadcastTimeout", 20 * 60)\
              .config("spark.sql.crossJoin.enabled", True)\
              .config("odps.exec.dynamic.partition.mode", "nonstrict")\
              .config("spark.sql.catalogImplementation", "odps")\
              .getOrCreate()
      
      def is_number(s):
          try:
              float(s)
              return True
          except ValueError:
              pass
      
          try:
              import unicodedata
              unicodedata.numeric(s)
              return True
          except (TypeError, ValueError):
              pass
      
          return False
      
      print(is_number('foo'))
      print(is_number('1'))
      print(is_number('1.3'))
      print(is_number('-1.37'))
      print(is_number('1e3'))

   2. 保存并提交资源。

2. 在已创建的ODPS Spark节点中，根据配置项说明配置节点参数和调度配置参数，并保存提交节点。

   配置项	说明
   spark版本	Spark2.x
   语言	Python
   选择主python资源	在下拉列表中选择上述已创建的python资源spark_is_number.py

3. 进入开发环境的运维中心，执行补数据，具体操作请参见执行补数据并查看补数据实例（新版）。

   说明

   由于数据开发中的ODPS Spark节点没有运行入口，因此您需要在开发环境的运维中心执行Spark任务。

4. 查看返回结果。

   待补数据实例运行成功后，进入其运行日志的tracking URL中查看运行结果，如下：

   False
   True
   True
   True
   True

编辑代码：进阶示例

更多场景的Spark on MaxCompute任务开发，请参考：

• java/scala示例：Spark-1.x示例

• java/scala示例：Spark-2.x示例

• Python示例：PySpark开发示例

• 场景：Spark访问VPC实例

• 场景：Spark访问OSS