DataWorks:EMR PySpark node

Step 1: Upload dependent resources

Upload your custom Python file to the Resource Management module so the node can reference it at runtime.

1. In Data Development, go to the resource management page and click Create Resource. Select EMR File as the resource type and set the resource Name. For detailed steps, see EMR resources and functions.

2. Click Re-Upload to upload the example utils.py file. This file defines the Monte Carlo sampling logic executed within each Spark task.

3. Select the Storage path, Connection, and Resource Groups, then click Save.

   

Step 2: Write the Python code

In the Python code editor (upper pane), write the following code. The program divides the workload across partitions, runs Monte Carlo sampling in parallel on worker nodes, and then aggregates the results to estimate pi.

##@resource_reference{"utils.py"}
from pyspark.sql import SparkSession
from utils import estimate_pi_in_task
import sys

def main():
    # Create a SparkSession
    spark = SparkSession.builder \
        .appName("EstimatePi") \
        .getOrCreate()

    sc = spark.sparkContext

    # Total number of samples
    total_samples = int(sys.argv[1])
    num_partitions = ${test1}

    # Number of samples per partition
    samples_per_partition = total_samples // num_partitions

    # Create an RDD where each partition executes estimate_pi_in_task once
    rdd = sc.parallelize(range(num_partitions), num_partitions)

    # Map each partition to execute the sampling task
    inside_counts = rdd.map(lambda _: estimate_pi_in_task(samples_per_partition))

    # Aggregate the results from all partitions
    total_inside = inside_counts.sum()
    pi_estimate = 4.0 * total_inside / total_samples

    print(f"Total samples: {total_samples}")
    print(f"Samples inside the circle: {total_inside}")
    print(f"Estimated value of Pi: {pi_estimate:.6f}")

    spark.stop()

if __name__ == "__main__":
    main()

The code uses two types of parameters:

Step 3: Write the spark-submit command

In the command editor (lower pane), write the spark-submit command. The required parameters differ depending on your cluster type.

For semi-managed clusters (EMR compute resources) — uses the standard open-source Apache Spark spark-submit tool:

# Submit to a semi-managed EMR cluster (DataLake or Custom)
spark-submit \
  --master yarn \         # Specify YARN as the Spark cluster manager.
  --deploy-mode cluster \ # Specify the cluster deployment mode.
  --py-files utils.py \  # Explicitly declare the dependent Python resource file.
  emr_pyspark_test.py 10000  # Main entry file of the program.

For fully managed clusters (EMR Serverless Spark) — uses the Alibaba Cloud EMR Serverless spark-submit toolkit, which handles cluster management automatically:

# Submit to EMR Serverless Spark
spark-submit \
  --py-files utils.py \
  emr_pyspark_test.py 10000

For a full list of supported parameters:

• Semi-managed clusters: see the Apache Spark documentation on submitting applications

• EMR Serverless Spark: see Submit a job by using spark-submit