<jupyter_start><jupyter_text>Feature Transformation with a Amazon SageMaker Processing Job and SparkMLTypically a machine learning (ML) process consists of few steps. First, gathering data with various ETL jobs, then pre-processing the data, featurizing the dataset by incorporating standard techniques or prior knowledge.Often, distributed data processing frameworks such as Spark are used to pre-process data sets in order to prepare them for training. In this notebook we'll use Amazon SageMaker Processing, and leverage the power of Spark in a managed SageMaker environment to run our processing workload. ![](img/prepare_dataset_tfidf.png)![](img/processing.jpg)  Contents1. Setup1. Use Amazon SageMaker Processing Job to execute a Spark ML Transformation1. Setup Input Data1. Setup Output Data1. Build a Spark container for running the processing job1. Run the Processing Job using Amazon SageMaker1. Inspect the Processed Output Dta  SetupAdd the following policies to your SageMaker Execution Role:  * `EC2ContainerRegistry`* Permissions: `List`, `Read`, `Write` * Repository:  `amazon-reviews-spark-processor`Let's start by specifying:* The S3 bucket and prefixes that you use for training and model data. Use the default bucket specified by the Amazon SageMaker session.* The IAM role ARN used to give processing and training access to the dataset.<jupyter_code>import sagemaker
import boto3

sagemaker_session = sagemaker.Session()
role = sagemaker.get_execution_role()
bucket = sagemaker_session.default_bucket()
region = boto3.Session().region_name<jupyter_output><empty_output><jupyter_text>Using Amazon SageMaker Processing Job to Run a Spark ML Transformation Get the name of previous Scikit Processing Job to use to construct the input for our Spark ML Processing Job.  Build a Spark Docker Image to Run the Processing JobAn example Spark container is included in the `./container` directory of this example. The container handles the bootstrapping of all Spark configuration, and serves as a wrapper around the `spark-submit` CLI. At a high level the container provides:* A set of default Spark/YARN/Hadoop configurations* A bootstrapping script for configuring and starting up Spark master/worker nodes* A wrapper around the `spark-submit` CLI to submit a Spark applicationAfter the container build and push process is complete, use the Amazon SageMaker Python SDK to submit a managed, distributed Spark application that performs our dataset processing.Build the example Spark container.<jupyter_code>docker_repo = 'amazon-reviews-spark-processor'
docker_tag = 'latest'
!docker build -t $docker_repo:$docker_tag -f container/Dockerfile ./container<jupyter_output><empty_output><jupyter_text>Create an Amazon Elastic Container Registry (ECR) RepositoryThis will store the Spark container that we build.<jupyter_code>import boto3
account_id = boto3.client('sts').get_caller_identity().get('Account')
region = boto3.session.Session().region_name

image_uri = '{}.dkr.ecr.{}.amazonaws.com/{}:{}'.format(account_id, region, docker_repo, docker_tag)
print(image_uri)<jupyter_output><empty_output><jupyter_text>Create ECR repository and push docker image<jupyter_code>!$(aws ecr get-login --region $region --registry-ids $account_id --no-include-email)
!aws ecr describe-repositories --repository-names $docker_repo || aws ecr create-repository --repository-name $docker_repo
!docker tag $docker_repo:$docker_tag $image_uri
!docker push $image_uri<jupyter_output><empty_output><jupyter_text>Run the Job using Amazon SageMaker Processing JobsNext, use the Amazon SageMaker Python SDK to submit a processing job. Use the Spark container that was just built, and a SparkML script for processing in the job configuration. Review the Spark processing script.<jupyter_code>!pygmentize preprocess-spark-text-to-tfidf.py
from sagemaker.processing import ScriptProcessor

processor = ScriptProcessor(base_job_name='spark-amazon-reviews-processor',
                            image_uri=image_uri,
                            command=['/opt/program/submit'],
                            role=role,
                            instance_count=2, # instance_count needs to be > 1 or you will see the following error:  "INFO yarn.Client: Application report for application_ (state: ACCEPTED)"
                            instance_type='ml.r5.8xlarge',
                            env={'mode': 'python'})<jupyter_output><empty_output><jupyter_text>Start the Spark Processing Job_Notes on Invoking from Lambda:_* However, if we use the boto3 SDK (ie. with a Lambda), we need to copy the `preprocess-text-to-tfidf.py` file to S3 and specify the everything include --py-files, etc.* We would need to do the following before invoking the Lambda:     !aws s3 cp preprocess.py s3:///sagemaker/spark-preprocess-reviews-demo/code/preprocess.py     !aws s3 cp preprocess.py s3:///sagemaker/spark-preprocess-reviews-demo/py_files/preprocess.py* Then reference the s3:// above in the --py-files, etc.* See Lambda example code in this same project for more details._Notes on not using ProcessingInput and Output:_* Since Spark natively reads/writes from/to S3 using s3a://, we can avoid the copy required by ProcessingInput and ProcessingOutput (FullyReplicated or ShardedByS3Key) and just specify the S3 input and output buckets/prefixes._"* See https://github.com/awslabs/amazon-sagemaker-examples/issues/994 for issues related to using /opt/ml/processing/input/ and output/* If we use ProcessingInput, the data will be copied to each node (which we don't want in this case since Spark already handles this)  Setup Input Data<jupyter_code># Inputs
s3_input_data = 's3://{}/amazon-reviews-pds/tsv/'.format(bucket)
print(s3_input_data)
!aws s3 ls $s3_input_data<jupyter_output><empty_output><jupyter_text>Setup Output Data<jupyter_code>from time import gmtime, strftime
timestamp_prefix = strftime("%Y-%m-%d-%H-%M-%S", gmtime())

output_prefix = 'amazon-reviews-spark-processor-{}'.format(timestamp_prefix)
train_data_tfidf_output = 's3://{}/{}/output/tfidf-train'.format(bucket, output_prefix)
validation_data_tfidf_output = 's3://{}/{}/output/tfidf-validation'.format(bucket, output_prefix)
test_data_tfidf_output = 's3://{}/{}/output/tfidf-test'.format(bucket, output_prefix)

print(train_data_tfidf_output)
print(validation_data_tfidf_output)
print(test_data_tfidf_output)
from sagemaker.processing import ProcessingOutput

processor.run(code='preprocess-spark-text-to-tfidf.py',
              arguments=['s3_input_data', s3_input_data,
                         's3_output_train_data', train_data_tfidf_output,
                         's3_output_validation_data', validation_data_tfidf_output,
                         's3_output_test_data', test_data_tfidf_output,                         
              ],
              # We need this dummy output to allow us to call 
              #    ProcessingJob.from_processing_name() later 
              #    to describe the job and poll for Completed status
              outputs=[
                       ProcessingOutput(s3_upload_mode='EndOfJob',
                                        output_name='dummy-output',
                                        source='/opt/ml/processing/output')
              ],          
              logs=True,
              wait=False
)
from IPython.core.display import display, HTML

spark_processing_job_name = processor.jobs[-1].describe()['ProcessingJobName']

display(HTML('<b>Review <a href="https://console.aws.amazon.com/cloudwatch/home?region={}#logStream:group=/aws/sagemaker/ProcessingJobs;prefix={};streamFilter=typeLogStreamPrefix">CloudWatch Logs</a> After About 5 Minutes</b>'.format(region, spark_processing_job_name)))

from IPython.core.display import display, HTML

# This is different than the job name because we are not using ProcessingOutput's in this Spark ML case.
spark_processing_job_s3_output_prefix = output_prefix

display(HTML('<b>Review <a href="https://s3.console.aws.amazon.com/s3/buckets/{}/{}/?region={}&tab=overview">S3 Output Data</a> After The Spark Job Has Completed</b>'.format(bucket, spark_processing_job_s3_output_prefix, region)))<jupyter_output><empty_output><jupyter_text>Please Wait Until the Processing Job CompletesRe-run this next cell until the job status shows `Completed`.<jupyter_code>running_processor = sagemaker.processing.ProcessingJob.from_processing_name(processing_job_name=spark_processing_job_name,
                                                                            sagemaker_session=sagemaker_session)

processing_job_description = running_processor.describe()

processing_job_status = processing_job_description['ProcessingJobStatus']
print('\n')
print(processing_job_status)
print('\n')

print(processing_job_description)<jupyter_output><empty_output><jupyter_text>Inspect the Processed Output Dataset The next cells will not work properly until the job completes above.Take a look at a few rows of the transformed dataset to make sure the processing was successful.<jupyter_code>!aws s3 ls --recursive $train_data_tfidf_output/
!aws s3 ls --recursive $validation_data_tfidf_output/
!aws s3 ls --recursive $test_data_tfidf_output/<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.08472735172713448, 'fraction_numerical': 0.006082989354768629, 'mean_word_length': 4.321965317919076, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 46, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '93725', 'license': 'Apache-2.0', 'n_tokens_mistral': 2678, 'n_tokens_neox': 2541, 'n_words': 814, 'path': '06_prepare/archive/99_Unused_Prepare_Dataset_ProcessingJob_Spark_TFIDF.ipynb', 'repo_name': 'MarcusFra/workshop', 'size': '9206'}