E-MapReduce:Developing batch or streaming tasks

Prerequisites

A workspace is created. For more information, see Manage workspaces.

Procedure

1. Go to the Data Development page.

   1. Log on to the E-MapReduce console.

   2. In the left navigation pane, choose EMR Serverless > Spark.

   3. On the Spark page, click the target workspace name.

   4. On the EMR Serverless Spark page, click Development in the left navigation pane.

2. Create a task.

   1. On the Development tab, click the (New) icon.

   2. In the dialog box, enter a Name, select a batch or streaming task type, and click OK.

   3. In the upper-right corner, select a resource queue.

   4. In the editor for the new task, edit the parameters based on your task type.

      JAR

      Parameter	Description
      Main JAR Resource	The primary JAR package required to run the task. Workspace: Files that you have previously uploaded to the Artifacts page. OSS: A file stored in Alibaba Cloud Object Storage Service (OSS).
      Engine Version	The Spark version. For more information, see Engine versions.
      Main class	The main class specified when you submit the Spark task.
      Execution Parameters	Configuration items required during task runtime or custom parameters passed to the main class. Separate multiple parameters with spaces.
      Timeout	The maximum time allowed for this task to complete. If the task runs longer than this threshold, the system automatically stops the task. The default value is empty, which means no timeout limit is set.
      Network Connection	Select an existing network connection to access data sources in a VPC or external services. For more information about how to create a network connection, see Network connectivity between EMR Serverless Spark and other VPCs.
      Mount Integrated File Directory	This feature is disabled by default. To use this feature, add a file directory on the Artifacts page, on the Integrated File Directory tab. For more information, see Manage the integrated file directory. Enabling this feature mounts the managed file directory to the application, allowing it to directly read from and write to files in the directory. The mount operation consumes driver compute resources. The amount consumed is the greater of the following two values: Fixed resources: 0.3 vCPUs + 1 GB memory. Dynamic resources: 10% of the spark.driver resources (that is, 10% of the cores and memory of spark.driver). For example, if spark.driver is configured with 4 cores and 8 GB of memory, the dynamic resources are 0.4 vCPUs + 0.8 GB of memory. In this case, the actual consumed resources are max(0.3 vCPUs + 1 GB, 0.4 vCPUs + 0.8 GB), which is 0.4 vCPUs + 1 GB of memory. Note Mount scope: By default, the file directory is mounted only to the driver. To mount it to executors as well, enable Mount to Executor. Multiple directories: You can mount multiple integrated file directories. However, CPFS directories cannot be used together with other types. For example, you can mount multiple OSS and NAS directories together, but you cannot mount CPFS with OSS or NAS directories. Network requirements: When you mount a NAS or CPFS file directory, you must configure a network connection. The VPC of the network connection must be the same as the VPC of the NAS or CPFS mount target.
      Mount to Executor	Enabling this feature mounts the managed file directory to the application's executors, allowing them to directly read from and write to files in the directory. This mount operation consumes executor resources. The amount of resources consumed varies based on the usage of the mounted files.
      File Resources	When you submit the task, files specified by the --files parameter are copied to the working directory of the executor. This ensures that the Spark task can access these files at runtime. You can select files from either Workspace or OSS.
      Archive Resources	When you submit the task, files specified by the --archives parameter are decompressed and distributed to the archived objects on the executor. You can select archives from either Workspace or OSS.
      JAR Resources	When you submit the task, specify the required JAR dependency files using the --jars parameter. You can select JARs from either Workspace or OSS.
      spark.driver.cores	The number of CPU cores used by the driver in the Spark application.
      spark.driver.memory	The amount of memory available to the driver in the Spark application.
      spark.executor.cores	The number of virtual CPU cores used by each executor in the Spark application.
      spark.executor.memory	The amount of memory available to each executor in the Spark application.
      spark.executor.instances	The number of executors allocated by Spark.
      Dynamic Resource Allocation	Disabled by default. When enabled, configure the following parameters: Minimum Number of Executors: The default value is 2. Maximum Number of Executors: If spark.executor.instances is not set, the default value is 10.
      More Memory Configurations	spark.driver.memoryOverhead: The non-heap memory available for the driver. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.driver.memory). spark.executor.memoryOverhead: The non-heap memory available for each executor. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.executor.memory). spark.memory.offHeap.size: The amount of off-heap memory available to Spark. The default value is 1 GB. This parameter takes effect only when spark.memory.offHeap.enabled is set to true. When the Fusion engine is used, this feature is enabled by default with 1 GB of off-heap memory.
      Spark Configuration	Enter Spark configuration information. Separate key-value pairs with a space. For example, key value.
      Tags	Enter key-value pairs for tags. Tags help you manage tasks with more convenience and precision.

      PySpark

      Parameter	Description
      Main Python Resources	The primary Python file required to run the task. Workspace: A file that you have uploaded on the Artifacts page. OSS: A file stored in Object Storage Service (OSS).
      Engine Version	The Spark version. For more information, see Engine versions.
      Execution Parameters	Configuration items required during task runtime or custom parameters passed to the main class.
      Timeout	The maximum time allowed for this task to complete. If the task runs longer than this threshold, the system automatically stops the task. The default value is empty, which means no timeout limit is set.
      Environment	The resources required to run the task are pre-configured based on the selected environment.
      Network Connection	Select an existing network connection to access data sources in a VPC or external services. For more information about how to create a network connection, see Network connectivity between EMR Serverless Spark and other VPCs.
      Mount Integrated File Directory	This feature is disabled by default. To use this feature, add a file directory on the Artifacts page, on the Integrated File Directory tab. For more information, see Manage the integrated file directory. When enabled, this feature mounts the managed file directory to the task, allowing direct read and write access to files in the directory. The mount operation consumes driver compute resources. The amount consumed is the greater of the following two values: Fixed resources: 0.3 vCPUs + 1 GB memory. Dynamic resources: 10% of the spark.driver resources (that is, 10% of the cores and memory of spark.driver). For example, if spark.driver is configured with 4 cores and 8 GB of memory, the dynamic resources are 0.4 vCPUs + 0.8 GB of memory. In this case, the actual consumed resources are max(0.3 vCPUs + 1 GB, 0.4 vCPUs + 0.8 GB), which is 0.4 vCPUs + 1 GB of memory. Note Mount scope: By default, the file directory is mounted only to the driver. To mount it to executors as well, enable Mount to Executor. Multiple directories: You can mount multiple integrated file directories. However, CPFS directories cannot be used together with other types. For example, you can mount multiple OSS and NAS directories together, but you cannot mount CPFS with OSS or NAS directories. Network requirements: When you mount a NAS or CPFS file directory, you must configure a network connection. The VPC of the network connection must be the same as the VPC of the NAS or CPFS mount target.
      Mount to Executor	When enabled, this feature mounts the managed file directory to the task's executors, allowing them direct read and write access to files in the directory. This mount operation consumes executor resources. The percentage of consumed resources varies based on the usage of the mounted files.
      File Resources	A list of files to be distributed to all executor nodes in the cluster. For the resource type, you can select Workspace or OSS.
      Pyfiles Resources	When you submit the task, files specified by the --py-files parameter are distributed as Python dependency files. For the resource type, you can select Workspace or OSS.
      Archive Resources	When you submit the task, files specified by the --archives parameter are decompressed and distributed to the archived objects on the executor. For the resource type, you can select Workspace or OSS.
      JAR Resources	When you submit the task, specify the required JAR dependency files using the --jars parameter. For the resource type, you can select Workspace or OSS.
      spark.driver.cores	The number of CPU cores used by the driver in the Spark application.
      spark.driver.memory	The amount of memory available to the driver in the Spark application.
      spark.executor.cores	The number of virtual CPU cores used by each executor in the Spark application.
      spark.executor.memory	The amount of memory available to each executor in the Spark application.
      spark.executor.instances	The number of executors allocated by Spark.
      Dynamic Resource Allocation	Disabled by default. When enabled, configure the following parameters: Minimum Number of Executors: The default value is 2. Maximum Number of Executors: If spark.executor.instances is not set, the default value is 10.
      More Memory Configurations	spark.driver.memoryOverhead: The non-heap memory available for the driver. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.driver.memory). spark.executor.memoryOverhead: The non-heap memory available for each executor. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.executor.memory). spark.memory.offHeap.size: The amount of off-heap memory available to Spark. The default value is 1 GB. This parameter takes effect only when spark.memory.offHeap.enabled is set to true. When the Fusion engine is used, this feature is enabled by default with 1 GB of off-heap memory.
      Spark Configuration	Enter Spark configuration information. Separate key-value pairs with a space. For example, key value.
      Tags	Enter key-value pairs for tags. Tags help you manage tasks with more convenience and precision.

      SQL

      Parameter	Description
      SQL File	The file required when you submit the task. Workspace: A file from the Artifacts page. OSS: A file from Alibaba Cloud OSS.
      Engine Version	The Spark version. For more information, see Engine versions.
      Timeout	The maximum time allowed for this task to complete. If the task runs longer than this threshold, the system automatically stops the task. The default value is empty, which means no timeout limit is set.
      Network Connection	Select an existing network connection to access data sources in a VPC or external services. For more information about how to create a network connection, see Network connectivity between EMR Serverless Spark and other VPCs.
      Mount Integrated File Directory	This feature is disabled by default. To use this feature, add a file directory on the Artifacts page, on the Integrated File Directory tab. For more information, see Manage the integrated file directory. When enabled, the system mounts the managed file directory to the task, allowing it to read from and write to the directory. The mount operation consumes driver compute resources. The amount consumed is the greater of the following two values: Fixed resources: 0.3 vCPUs + 1 GB memory. Dynamic resources: 10% of the spark.driver resources (that is, 10% of the cores and memory of spark.driver). For example, if spark.driver is configured with 4 cores and 8 GB of memory, the dynamic resources are 0.4 vCPUs + 0.8 GB of memory. In this case, the actual consumed resources are max(0.3 vCPUs + 1 GB, 0.4 vCPUs + 0.8 GB), which is 0.4 vCPUs + 1 GB of memory. Note Mount scope: By default, the file directory is mounted only to the driver. To mount it to executors as well, enable Mount to Executor. Multiple directories: You can mount multiple integrated file directories. However, CPFS directories cannot be used together with other types. For example, you can mount multiple OSS and NAS directories together, but you cannot mount CPFS with OSS or NAS directories. Network requirements: When you mount a NAS or CPFS file directory, you must configure a network connection. The VPC of the network connection must be the same as the VPC of the NAS or CPFS mount target.
      Mount to Executor	When enabled, the system mounts the managed file directory to the task's Executors, allowing them to read from and write to the directory. The mount operation consumes Executor resources. The percentage of resources consumed varies with file usage.
      spark.driver.cores	The number of CPU cores used by the driver in the Spark application.
      spark.driver.memory	The amount of memory available to the driver in the Spark application.
      spark.executor.cores	The number of virtual CPU cores used by each executor in the Spark application.
      spark.executor.memory	The amount of memory available to each executor in the Spark application.
      spark.executor.instances	The number of executors allocated by Spark.
      Dynamic Resource Allocation	Disabled by default. When enabled, configure the following parameters: Minimum Number of Executors: The default value is 2. Maximum Number of Executors: If spark.executor.instances is not set, the default value is 10.
      More Memory Configurations	spark.driver.memoryOverhead: The non-heap memory available for the driver. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.driver.memory). spark.executor.memoryOverhead: The non-heap memory available for each executor. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.executor.memory). spark.memory.offHeap.size: The amount of off-heap memory available to Spark. The default value is 1 GB. This parameter takes effect only when spark.memory.offHeap.enabled is set to true. When the Fusion engine is used, this feature is enabled by default with 1 GB of off-heap memory.
      Spark Configuration	Enter Spark configuration information. Separate key-value pairs with a space. For example, key value.
      Tags	Enter key-value pairs for tags. Tags help you manage tasks with more convenience and precision.

      Spark Submit

      Parameter	Description
      Engine Version	The Spark version. For more information, see Engine versions.
      Script	Enter your Spark Submit script. The following code provides an example: --class org.apache.spark.examples.SparkPi \ --conf spark.executor.memory=2g \ oss://<YourBucket>/spark-examples_2.12-3.5.2.jar
      Timeout	The maximum time allowed for this task to complete. If the task runs longer than this threshold, the system automatically stops the task. The default value is empty, which means no timeout limit is set.
      Network Connection	Select an existing network connection to access data sources in a VPC or external services. For more information about how to create a network connection, see Network connectivity between EMR Serverless Spark and other VPCs.
      Mount Integrated File Directory	This feature is disabled by default. To use this feature, add a file directory on the Artifacts page, on the Integrated File Directory tab. For more information, see Manage the integrated file directory. When enabled, this feature mounts the managed file directory to the application, allowing it to directly read from and write to the directory. The mount operation consumes driver compute resources. The amount consumed is the greater of the following two values: Fixed resources: 0.3 vCPUs + 1 GB memory. Dynamic resources: 10% of the spark.driver resources (that is, 10% of the cores and memory of spark.driver). For example, if spark.driver is configured with 4 cores and 8 GB of memory, the dynamic resources are 0.4 vCPUs + 0.8 GB of memory. In this case, the actual consumed resources are max(0.3 vCPUs + 1 GB, 0.4 vCPUs + 0.8 GB), which is 0.4 vCPUs + 1 GB of memory. Note Mount scope: By default, the file directory is mounted only to the driver. To mount it to executors as well, enable Mount to Executor. Multiple directories: You can mount multiple integrated file directories. However, CPFS directories cannot be used together with other types. For example, you can mount multiple OSS and NAS directories together, but you cannot mount CPFS with OSS or NAS directories. Network requirements: When you mount a NAS or CPFS file directory, you must configure a network connection. The VPC of the network connection must be the same as the VPC of the NAS or CPFS mount target.
      Mount to Executor	When enabled, this feature mounts the managed file directory to the application's executors, allowing them to directly read from and write to the directory. The amount of executor resources consumed depends on how the mounted files are used.
      spark.driver.cores	The number of CPU cores used by the driver in the Spark application.
      spark.driver.memory	The amount of memory available to the driver in the Spark application.
      spark.executor.cores	The number of virtual CPU cores used by each executor in the Spark application.
      spark.executor.memory	The amount of memory available to each executor in the Spark application.
      spark.executor.instances	The number of executors allocated by Spark.
      Dynamic Resource Allocation	Disabled by default. When enabled, configure the following parameters: Minimum Number of Executors: The default value is 2. Maximum Number of Executors: If spark.executor.instances is not set, the default value is 10.
      More Memory Configurations	spark.driver.memoryOverhead: The non-heap memory available for the driver. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.driver.memory). spark.executor.memoryOverhead: The non-heap memory available for each executor. If this parameter is not set, Spark automatically allocates a value based on the default, which is max(384 MB, 10% * spark.executor.memory). spark.memory.offHeap.size: The amount of off-heap memory available to Spark. The default value is 1 GB. This parameter takes effect only when spark.memory.offHeap.enabled is set to true. When the Fusion engine is used, this feature is enabled by default with 1 GB of off-heap memory.
      Spark Configuration	Enter Spark configuration information. Separate key-value pairs with a space. For example, key value.
      Tags	Enter key-value pairs for tags. Tags help you manage tasks with more convenience and precision.

   5. (Optional) On the right side of the task development page, click the Version Information tab to view or compare versions.

3. Run and publish the task.

   1. Click Run.

      To view task details, go to the Execution Records area at the bottom after the task has run, and click Details in the Actions column.

   2. In the upper-right corner, click Publish.

   3. In the Publish dialog box, enter Remarks and click OK.

Related documents

• Quick Start for JAR development

• Quick Start for Spark Submit development

• Quick Start for SparkSQL development

• Quick Start for PySpark development

• Submit a PySpark streaming task using Serverless Spark

FAQ

Q1: How do I set an automatic retry policy for failed tasks?

To improve the fault tolerance of streaming tasks, you can set an automatic retry policy using the following two Spark configuration items:

spark.emr.serverless.streaming.fail.retry.interval 60    # Retry interval: 60 seconds
spark.emr.serverless.streaming.fail.retry.time 3        # Maximum retries: 3