text_plain_output_1.png
from pyspark.ml  import  Pipeline

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler

from pyspark.sql import SparkSession

from pyspark.sql import SparkSession
my_spark = SparkSession.builder.getOrCreate()

file_path = '../input/flights.csv'
flights = my_spark.read.csv(file_path, header=True)
flights.createOrReplaceTempView('flights')

flights = flights.withColumn('duration_hrs', flights.air_time / 60)

flights.toPandas().shape[0]

file_path = '../input/planes.csv'
planes = my_spark.read.csv(file_path, header=True)
planes = planes.withColumnRenamed('year', 'plane_year')
model_data = flights.join(planes, on='tailnum', how='leftouter')

model_data = model_data.withColumn('arr_delay', model_data.arr_delay.cast('integer'))
model_data = model_data.withColumn('air_time', model_data.air_time.cast('integer'))
model_data = model_data.withColumn('month', model_data.month.cast('integer'))
model_data = model_data.withColumn('plane_year', model_data.plane_year.cast('integer'))

model_data = model_data.withColumn('plane_age', model_data.year - model_data.plane_year)

model_data = model_data.withColumn('is_late', model_data.arr_delay > 0)
model_data = model_data.withColumn('label', model_data.is_late.cast('integer'))
model_data = model_data.filter('arr_delay is not NULL and dep_delay is not NULL and air_time is not NULL and plane_year is not NULL')

from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
carr_indexer = StringIndexer(inputCol='carrier', outputCol='carrier_index')
carr_encoder = OneHotEncoder(inputCol='carrier_index', outputCol='carrier_fact')

dest_indexer = StringIndexer(inputCol='dest', outputCol='dest_index')
dest_encoder = OneHotEncoder(inputCol='dest_index', outputCol='dest_fact')

vec_assembler = VectorAssembler(inputCols=['month', 'air_time', 'carrier_fact', 'dest_fact', 'plane_age'], outputCol='features')

from pyspark.ml import Pipeline
flights_pipe = Pipeline(stages=[dest_indexer, dest_encoder, carr_indexer, carr_encoder, vec_assembler])

piped_data = flights_pipe.fit(model_data).transform(model_data)

piped_data.toPandas().head(3)