EMR PySpark node

Step 1: Upload resources

Upload your custom Python file (in this example, utils.py) to the Resource Management module in DataWorks. After you upload the file, you can reference this resource in your node. This Python file defines the core logic of the Monte Carlo sampling simulation within a single Spark task.

1. For detailed instructions on uploading resources, see EMR resources and functions. On the Resource Management page in DataStudio, click Create Resource, select EMR File as the resource type, and configure the resource Name.

2. Click Re-Upload to upload the example utils.py file.

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

   

Step 2: Write the Python code

In the Python code editor, write the following code. This example program estimates the value of Pi (π) by using parallel computing on a cluster. The program divides the computation into subtasks, distributes them to compute nodes, and then collects and aggregates the results to calculate the final estimate for 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 runs estimate_pi_in_task once
    rdd = sc.parallelize(range(num_partitions), num_partitions)

    # Map each partition to run 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 within the circle: {total_inside}")
    print(f"Estimated value of Pi: {pi_estimate:.6f}")

    spark.stop()

if __name__ == "__main__":
    main()

The following table describes the parameters used in the script.

Step 3: Write the spark-submit command

In the Submit command editor, write the following code. This command uses the spark-submit tool to package and submit the Python application to an EMR cluster for execution.

The spark-submit command and its parameters differ significantly between EMR compute resources (semi-managed clusters) and EMR Serverless Spark compute resources (fully-managed clusters). The following table compares the command usage for both resource types.

spark-submit \                          
 --master yarn \            # Specify YARN as the Spark cluster manager.
 --deploy-mode cluster \    # Specify cluster as the deployment mode.
 --py-files utils.py \      # Explicitly declare the dependent Python resource file.
 emr_pyspark_test.py 10000  # Main application entry file.

spark-submit \                                       
 --py-files utils.py \                    
 emr_pyspark_test.py 10000  

Next steps

• Configure node scheduling: If you need to run a node periodically, configure its Scheduling Policy in the Scheduling Settings panel on the right.

• Publish a node: To run a task in the production environment, click the icon to publish the node. A node runs on schedule only after it is published to the production environment.

• Task O&M: After a task is published, you can monitor the status of its periodic runs in the Operation Center. For more information, see Get started with Operation Center.