$ docker pull jupyter/pyspark-notebook

$ docker run --name spark_test -it -p 8888:8888 -v $PWD:/home/jovyan/work jupyter/pyspark-notebook

import pandas as pd

import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('pandasToSparkDF').getOrCreate()

from pyspark.sql.functions import mean, col, split, regexp_extract, when, lit

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, VectorAssembler, QuantileDiscretizer
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# データ読み込み
titanic_df = spark.read.csv('./titanic/train.csv', header='True', inferSchema='True')

# 欠損値対応
titanic_df = titanic_df.na.fill({"Embarked" : 'S'})

# 不要のカラムを落とす
titanic_df = titanic_df.drop("Cabin")

# 定数による列追加
titanic_df = titanic_df.withColumn('Alone', lit(0))

# 条件による値の挿入
titanic_df = titanic_df.withColumn("Alone", when(titanic_df["Family_Size"] == 0, 1).otherwise(titanic_df["Alone"]))

# ラベルエンコード
indexers = [StringIndexer(inputCol=column, outputCol=column+"_index").fit(titanic_df) for column in ["Sex", "Embarked", "Initial"]]
pipeline = Pipeline(stages=indexers)
titanic_df = pipeline.fit(titanic_df).transform(titanic_df)

# テストスプリット
trainingData, testData = feature_vector.randomSplit([0.8, 0.2], seed=9)

# 学習
lr = LogisticRegression(labelCol="Survived", featuresCol="features")
lrModel = lr.fit(trainingData)

# 推論
lr_prediction = lrModel.transform(testData)
lr_prediction.select("prediction", "Survived", "features").show()

# 評価
evaluator = MulticlassClassificationEvaluator(labelCol="Survived", predictionCol="prediction", metricName="accuracy")
lr_accuracy = evaluator.evaluate(lr_prediction)

print("Accuracy of LogisticRegression is = %g"% (lr_accuracy))
print("Test Error of LogisticRegression = %g " % (1.0 - lr_accuracy))