E-MapReduce:PySpark streaming on EMR Serverless Spark

Step 1: Create a Dataflow cluster and produce messages

1. On the EMR on ECS page, create a real-time Dataflow cluster that includes the Kafka service. For more information, see Create a cluster.

2. Log on to the master node of the EMR on ECS cluster. For more information, see Log on to a cluster.

3. Run the following command to change the directory:

   cd /var/log/emr/taihao_exporter

4. Run the following command to create a topic:

   # Create a topic named taihaometrics with 10 partitions and a replication factor of 2.
   kafka-topics.sh --partitions 10 --replication-factor 2 --bootstrap-server core-1-1:9092 --topic taihaometrics --create

5. Run the following command to send messages:

   # Use kafka-console-producer to send messages to the taihaometrics topic.
   tail -f metrics.log | kafka-console-producer.sh --broker-list core-1-1:9092 --topic taihaometrics

Step 2: Create a network connection

1. Go to the Network Connection page.

   1. In the left-side navigation pane of the EMR console, choose EMR Serverless > Spark.

   2. On the Spark page, click the name of your target workspace.

   3. On the EMR Serverless Spark page, click Normal Network Connection in the left-side navigation pane.

2. On the Normal Network Connection page, click Create Network Connection.

3. In the Create Network Connection dialog box, configure the following parameters and click OK.

   Parameter	Description
   Name	Enter a name for the connection. For example, connection_to_emr_kafka.
   VPC	Select the VPC where your EMR on ECS cluster is deployed. If no VPC is available, click Create VPC to go to the VPC console and create a VPC. For more information, see Create and manage a VPC.
   vSwitch	Select the vSwitch that is in the same VPC as your EMR on ECS cluster. If no vSwitch is available in the current zone, click vSwitch to go to the VPC console and create a vSwitch. For more information, see Create and manage a vSwitch.

   When the Status is Succeeded, the network connection has been created.

Step 3: Add a security group rule

1. Obtain the vSwitch CIDR block for the cluster nodes.

   On the Nodes page, click a node group name to find the associated vSwitch. Then, log on to the VPC console and find the vSwitch's CIDR block on the vSwitch page.

   

2. Add a security group rule.

   1. On the Clusters page, click the ID of the target cluster.

   2. On the Basic Information page, click the link next to Cluster Security Group.

   3. On the Security Group Details page, in the Rules section, click Add Rule. Configure the following parameters and click OK.

      Parameter	Description
      Source	Enter the vSwitch CIDR block from the previous step. Important Do not set this to 0.0.0.0/0, as it exposes the cluster to external access.
      Destination (current instance)	Enter port 9092.

Step 4: Upload JAR packages to OSS

Extract kafka.zip and upload all JAR packages in the archive to OSS. For more information, see Simple upload.

Step 5: Upload the resource file

1. On the EMR Serverless Spark page, click Artifacts in the left-side navigation pane.

2. On the Artifacts page, click Upload File.

3. In the Upload File dialog box, click the upload area and select the pyspark_ss_demo.py file.

Step 6: Create and start a streaming job

1. On the EMR Serverless Spark page, click Development in the left-side navigation pane.

2. On the Development tab, click the icon.

3. Enter a name, select Application (Streaming) > PySpark as the job type, and then click OK.

4. In the new development tab, configure the following parameters, accept the defaults for the others, and then click Save.

   Parameter	Description
   Main Python Resources	Select the pyspark_ss_demo.py file that you uploaded on the Resource Upload page in the previous step.
   Engine Version	Select the Spark version. For more information, see Engine versions.
   Execution Parameters	Enter the internal IP address of the cluster's core-1-1 node. You can find this IP on the Nodes page, within the Core node group.
   Spark Configuration	Specify the Spark configurations. The following is an example. spark.jars oss://path/to/commons-pool2-2.11.1.jar,oss://path/to/kafka-clients-2.8.1.jar,oss://path/to/spark-sql-kafka-0-10_2.12-3.3.1.jar,oss://path/to/spark-token-provider-kafka-0-10_2.12-3.3.1.jar spark.emr.serverless.network.service.name connection_to_emr_kafka Note spark.jars: The OSS paths of the required external JAR packages. Replace the example paths with the OSS paths to the JARs you uploaded in Step 4. spark.emr.serverless.network.service.name: The name of the network connection. Replace the example value with the name of your network connection from Step 2.

5. Click Publish.

6. In the Publish dialog box, click OK.

7. Start the streaming job.

   1. Click Go to O&M.

   2. Click START.

Step 7: View the logs

1. Click the Log Exploration tab.

2. On the Log Exploration tab, view application execution details and results.