<jupyter_start><jupyter_text>Transformations* Drop date* Encode parameter* Encode location* Scale value* Drop unit* Encode city* Drop attribution* Drop averaging period* Drop coordinates* Encode country* Encode source name* Encode source type* Convert mobile to integer* Featurize aggdate* Add label for good/bad air quality<jupyter_code># Drop columns
df = df.drop('date').drop('unit').drop('attribution').drop('averagingperiod').drop('coordinates')
df.head(3)
# Mobile field to int
from pyspark.sql.functions import col
df = df.withColumn("ismobile",col("mobile").cast(IntegerType())).drop('mobile')
df.head(3)
# scale value
from pyspark.ml.feature import StandardScaler

value_assembler = VectorAssembler(inputCols=["value"], 
                            outputCol="value_vec")
value_scaler = StandardScaler(inputCol="value_vec", outputCol="value_scaled")
value_pipeline = Pipeline(stages=[value_assembler, value_scaler])
value_model = value_pipeline.fit(df)
xform_df = value_model.transform(df)
xform_df.head(3)

# featurize date
from pyspark.sql.functions import unix_timestamp, to_date
xform_df = xform_df.withColumn('aggdt', 
                   to_date(unix_timestamp(col('aggdate'), 'yyyy-MM-dd').cast("timestamp")))
xform_df.head(3)
from pyspark.sql.functions import year, month, quarter, date_format
xform_df = xform_df.withColumn('year',year(xform_df.aggdt)).withColumn('month',month(xform_df.aggdt)).withColumn('quarter',quarter(xform_df.aggdt))
xform_df = xform_df.withColumn("day", date_format(col("aggdt"), "d"))
xform_df.head(3)
# Automatically assign good/bad labels

from pyspark.sql.functions import udf

def isBadAir(v, p):
    if p == 'pm10':
        if v > 50:
            return 1
        else:
            return 0
    elif p == 'pm25':
        if v > 25:
            return 1
        else:
            return 0
    elif p == 'so2':
        if v > 20:
            return 1
        else:
            return 0
    elif p == 'no2':
        if v > 200:
            return 1
        else:
            return 0
    elif p == 'o3':
        if v > 100:
            return 1
        else:
            return 0
    else:
        return 0

isBadAirUdf = udf(isBadAir, IntegerType())

xform_df = xform_df.withColumn('isBadAir', isBadAirUdf('value', 'parameter'))
xform_df.head(3)<jupyter_output><empty_output><jupyter_text>Categorical encodingsFor low-cardinality features, we will use one-hot encoding.  For others we will use index (ordinal) encoding, in the expectation that we will calculate embeddings later on.<jupyter_code>from pyspark.ml.feature import OneHotEncoderEstimator

parameter_indexer = StringIndexer(inputCol="parameter", outputCol="indexed_parameter")
location_indexer = StringIndexer(inputCol="location", outputCol="indexed_location")
city_indexer = StringIndexer(inputCol="city", outputCol="indexed_city")
country_indexer = StringIndexer(inputCol="country", outputCol="indexed_country")
sourcename_indexer = StringIndexer(inputCol="sourcename", outputCol="indexed_sourcename")
sourcetype_indexer = StringIndexer(inputCol="sourcetype", outputCol="indexed_sourcetype")
enc_est = OneHotEncoderEstimator(inputCols=["indexed_parameter"],
                                 outputCols=["vec_parameter"])
enc_pipeline = Pipeline(stages=[parameter_indexer, location_indexer, city_indexer, country_indexer, sourcename_indexer, sourcetype_indexer, enc_est])
enc_model = enc_pipeline.fit(xform_df)
enc_df = enc_model.transform(xform_df)
enc_df.head(3)
param_cols = enc_df.schema.fields[17].metadata['ml_attr']['vals']
param_cols<jupyter_output><empty_output><jupyter_text>Split and save<jupyter_code>enc_df.printSchema()
final_df = enc_df.drop('parameter').drop('location').drop('city').drop('country').drop('sourcename').drop('sourcetype').drop('aggdate') \
    .drop('value_vec').drop('aggdt').drop('indexed_parameter')
#final_df.head(3)

from pyspark.sql.types import DoubleType

firstelement=udf(lambda v:str(v[0]),StringType())
final_df = final_df.withColumn('value_str', firstelement('value_scaled'))
final_df = final_df.withColumn("value",final_df.value_str.cast(DoubleType())).drop('value_str').drop('value_scaled')
#final_df.head(3)
from pyspark.ml.linalg import Vectors

def extract(row):
    return (row.value, row.ismobile, row.year, row.month, row.quarter, row.day, row.isBadAir, 
            row.indexed_location, row.indexed_city, row.indexed_sourcename, 
            row.indexed_sourcetype) + tuple(row.vec_parameter.toArray().tolist())

final_df = final_df.rdd.map(extract).toDF(["value", "ismobile", "year", "month", "quarter", "day", "isBadAir",
                               "location", "city", "sourcename", "sourcetype"] + param_cols[:-1])
final_df.head(3)
(train_df, validation_df, test_df) = final_df.randomSplit([0.7, 0.2, 0.1])
train_df.write.option("header",True).csv("s3://rdtest-data/prepared/train/")
test_df.write.option("header",True).csv("s3://rdtest-data/prepared/test/")
validation_df.write.option("header",True).csv("s3://rdtest-data/prepared/validation/")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13659793814432988, 'fraction_numerical': 0.010507533703409993, 'mean_word_length': 4.8189158016147635, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 18, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '155098', 'license': 'MIT', 'n_tokens_mistral': 1732, 'n_tokens_neox': 1681, 'n_words': 305, 'path': 'Chapter04/spark-example.ipynb', 'repo_name': 'bdonkey/Amazon-SageMaker-Best-Practices', 'size': '5044'}