<jupyter_start><jupyter_text>PySpark - Titanic UDFs  1. Load data<jupyter_code># load and cache data
df = sqlContext.read\
    .format("com.databricks.spark.csv")\
    .options(header = True)\
    .load("data/titanic_train.csv")\
    .cache()
# check DF column schema
df.printSchema()<jupyter_output>root
 |-- PassengerId: string (nullable = true)
 |-- Survived: string (nullable = true)
 |-- Pclass: string (nullable = true)
 |-- Name: string (nullable = true)
 |-- Sex: string (nullable = true)
 |-- Age: string (nullable = true)
 |-- SibSp: string (nullable = true)
 |-- Parch: string (nullable = true)
 |-- Ticket: string (nullable = true)
 |-- Fare: string (nullable = true)
 |-- Cabin: string (nullable = true)
 |-- Embarked: string (nullable = true)<jupyter_text>2. Clean data<jupyter_code>from pyspark.sql.functions import col
# select columns that we'll need, and cast to appropriate schema types
df = df.select(
    col("Survived").cast("double"),
    col("Pclass").cast("float"),
    col("Name"),
    col("Sex"),
    col("Age").cast("float"),
    col("SibSp").cast("float"),
    col("Parch"),
    col("Ticket"),
    col("Fare").cast("float"),
    col("Embarked"),
)
from pyspark.sql.functions import count
# count missing values in each column
def countNull(df, var):
    return df.where(df[var].isNull()).count()

allCols = df.columns
{var: countNull(df,var) for var in allCols}
# fill missing "Age" values with average
age_mean = df.groupBy().mean("Age").first()[0]
# fill missing "Embarked" values with average
embarked_mode = df.groupBy("Embarked").count().collect()[-1][0].encode("ascii")
df = df.fillna({"Age": age_mean, "Embarked": embarked_mode})
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
# compose another column with name salutations
getSalutation = udf(lambda name: name.split('.')[0].split(',')[1].strip(), StringType())
df = df.withColumn("Title", getSalutation(df["Name"]))
df.select('Name','Title').show(3)<jupyter_output>+--------------------+-----+
|                Name|Title|
+--------------------+-----+
|Braund, Mr. Owen ...|   Mr|
|Cumings, Mrs. Joh...|  Mrs|
|Heikkinen, Miss. ...| Miss|
+--------------------+-----+
only showing top 3 rows<jupyter_text>3. Featurization<jupyter_code># numerical variables
numVars = ["Age", "SibSp", "Parch", "Fare"]
# categorical variables
catVars = ["Pclass", "Sex", "Embarked", "Title", "Survived"]
from pyspark.ml.feature import StringIndexer
from pyspark.ml import Pipeline

# featurize categorical cols: one hot encoding
indexers = [
    StringIndexer(inputCol=col, outputCol="{0}_indx".format(col))
    for col in catVars
]
# add featurize stages to pipeline
pipeline = Pipeline(stages=indexers)
df_indexed = pipeline.fit(df).transform(df)
from pyspark.sql.types import Row
from pyspark.ml.linalg import Vectors

# convert features to vectors
cat_cols_indexed = [col+"_indx" for col in catVars]
feature_cols = cat_cols_indexed + numVars

# make label col the first column in DF
feature_cols.remove("Survived_indx")
label_col = ["Survived_indx"]
df_indexed = df_indexed[label_col+feature_cols]

# map features to DenseVector
row = Row('label', 'features')
lb_ft = df_indexed.rdd.map(lambda r: (row(r[0], Vectors.dense(r[1:])))).toDF()
lb_ft.show(5)<jupyter_output>+-----+--------------------+
|label|            features|
+-----+--------------------+
|  0.0|[0.0,0.0,0.0,0.0,...|
|  1.0|[1.0,1.0,1.0,2.0,...|
|  1.0|[0.0,1.0,0.0,1.0,...|
|  1.0|[1.0,1.0,0.0,2.0,...|
|  0.0|[0.0,0.0,0.0,0.0,...|
+-----+--------------------+
only showing top 5 rows
{'fraction_non_alphanumeric': 0.21528948847667229, 'fraction_numerical': 0.018268690275435637, 'mean_word_length': 3.592258064516129, 'pattern_counts': {'":': 2, '<': 10, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '438939', 'license': 'MIT', 'n_tokens_mistral': 1311, 'n_tokens_neox': 1289, 'n_words': 287, 'path': '_archived/titanic1_udf.ipynb', 'repo_name': 'Gilbertly/notebooks', 'size': '3558'}