code: def test_sagemaker_pyspark_multinode(
      spark_py_processor, sagemaker_session, configuration
  ):
      """Test that basic multinode case works on 32KB of data"""
      bucket = spark_py_processor.sagemaker_session.default_bucket()
      timestamp = datetime.now().isoformat()
      output_data_uri = f"s3://{bucket}/spark/output/sales/{timestamp}"
      spark_event_logs_key_prefix = f"spark/spark-events/{timestamp}"
      spark_event_logs_s3_uri = (
          f"s3://{bucket}/{spark_event_logs_key_prefix}"
      )
  
      with open(
          os.path.join(SPARK_PATH, "files", "data.jsonl")
      ) as data:
          body = data.read()
          input_data_uri = f"s3://{bucket}/spark/input/data.jsonl"
          S3Uploader.upload_string_as_file_body(
              body=body,
              desired_s3_uri=input_data_uri,
              sagemaker_session=sagemaker_session,
          )
  
      spark_py_processor.run(
          submit_app=os.path.join(
              SPARK_PATH,
              "code",
              "python",
              "hello_py_spark",
              "hello_py_spark_app.py",
          ),
          submit_py_files=[
              os.path.join(
                  SPARK_PATH,
                  "code",
                  "python",
                  "hello_py_spark",
                  "hello_py_spark_udfs.py",
              )
          ],
          arguments=[
              "--input",
              input_data_uri,
              "--output",
              output_data_uri,
          ],
          configuration=configuration,
          spark_event_logs_s3_uri=spark_event_logs_s3_uri,
          wait=False,
      )
      processing_job = spark_py_processor.latest_job
  
      s3_client = boto3.client(
          "s3",
          region_name=spark_py_processor.sagemaker_session.boto_region_name,
      )
  
      file_size = 0
      latest_file_size = None
      updated_times_count = 0
      time_out = time.time() + 900
      while not processing_job_not_fail_or_complete(
          sagemaker_session.sagemaker_client, processing_job.job_name
      ):
          response = s3_client.list_objects(
              Bucket=bucket, Prefix=spark_event_logs_key_prefix
          )
          if "Contents" in response:
              # somehow when call list_objects the first file size is always 0, this for loop
              # is to skip that.
              for event_log_file in response["Contents"]:
                  if event_log_file["Size"] != 0:
                      latest_file_size = event_log_file["Size"]
  
          # update the file size if it increased
          if latest_file_size and latest_file_size > file_size:
              updated_times_count += 1
              file_size = latest_file_size
  
          if time.time() > time_out:
              raise RuntimeError("Timeout")
  
          time.sleep(20)
  
      # verify that spark event logs are periodically written to s3
      assert file_size != 0
  
      output_contents = S3Downloader.list(
          output_data_uri, sagemaker_session=sagemaker_session
      )
      assert len(output_contents) != 0
  
imports: from __future__ import absolute_import
  import re
  import os
  import subprocess
  import time
  import urllib3
  from datetime import datetime
  from urllib3.util.retry import Retry
  import boto3
  import pytest
  from sagemaker.s3 import S3Downloader, S3Uploader
  from sagemaker.spark.processing import PySparkProcessor, SparkJarProcessor
  from tests.integ import DATA_DIR
  from unittest.case import TestCase
label_desc: Use image node_id 8 for calling a global function with example usage: test_sagemaker_pyspark_multinode(spark_py_processor, sagemaker_session, configuration) without return types