Introduction to Pyspark

Connect to Python

Creating the connection is as simple as creating an instance of the SparkContext class. The class constructor takes a few optional arguments that allow you to specify the attributes of the cluster you're connecting to.

# Verify SparkContext
print(sc)

# Print Spark version
print(sc.version)

To start working with Spark DataFrames, you first have to create a SparkSession object from your SparkContext. You can think of the SparkContext as your connection to the cluster and the SparkSession as your interface with that connection.

Create a SparkSession

# Import SparkSession from pyspark.sql
from pyspark.sql import SparkSession

# Create my_spark
my_spark = SparkSession.builder.getOrCreate()

# Print the tables in the catalog
print(spark.catalog.listTables())

SparkSession has an attribute called catalog which lists all the data inside the cluster. This attribute has a few methods for extracting different pieces of information.

One of the most useful is the .listTables()method, which returns the names of all the tables in your cluster as a list.

Pandas Dataframe <-> Spark Dataframe

.createDataFrame() method takes a pandas DataFrame and returns a Spark DataFrame. .toPandas()does the opposite.

# Convert the results to a pandas DataFrame
pd_counts = flight_counts.toPandas()

Spark Dataframe的内容存在locally，不在SparkSession catalog. This means that you can use all the Spark DataFrame methods on it, but you can't access the data in other contexts. 比如不能直接在spark dataframe上.sql() ，必须.createTempView() 或者 .createOrReplaceTempView()。

# Create pd_temp
pd_temp = pd.DataFrame(np.random.random(10))

# Create spark_temp from pd_temp
spark_temp = spark.createDataFrame(pd_temp)

# Examine the tables in the catalog
print(spark.catalog.listTables())

# Add spark_temp to the catalog
spark_temp.createOrReplaceTempView("temp")

# Examine the tables in the catalog again
print(spark.catalog.listTables())

比如上面这个代码，第一次tempview的listtable是[]，第二次是[Table(name='temp', database=None, description=None, tableType='TEMPORARY', isTemporary=True)]

withColumn

Spark DataFrame is immutable. This means that it can't be changed, and so columns can't be updated in place.

Thus, all these methods return a new DataFrame. To overwrite the original DataFrame you must reassign the returned DataFrame using the method like so:

df = df.withColumn("newCol", df.oldCol + 1)

# Create the DataFrame flights
flights = spark.table("flights")

# Add duration_hrs
flights = flights.withColumn("duration_hrs", flights.air_time/60)

Filter

The .filter() method takes either an expression that would follow the WHERE clause of a SQL expression as a string, or a Spark Column of boolean (True/False) values.

For example, the following two expressions will produce the same output:

# Filter flights by passing a string
long_flights1 = flights.filter("distance > 1000")

# Filter flights by passing a column of boolean values
long_flights2 = flights.filter(flights.distance > 1000)

Select

These arguments can either be the column name as a string (one for each column) or a column object (using the df.colName syntax).

The difference between .select() and .withColumn() methods is that .select() returns only the columns you specify, while .withColumn()returns all the columns of the DataFrame in addition to the one you defined.

# Select the first set of columns
selected1 = flights.select("tailnum", "origin", "dest")

# Select the second set of columns
temp = flights.select(flights.origin, flights.dest, flights.carrier)

# Define first filter
filterA = flights.origin == "SEA"

# Define second filter
filterB = flights.dest == "PDX"

# Filter the data, first by filterA then by filterB
selected2 = temp.filter(filterA).filter(filterB)

When you're selecting a column using the df.colName notation, you can perform any column operation and the .select() method will return the transformed column. For example,

flights.select((flights.air_time/60).alias("duration_hrs"))

The equivalent Spark DataFrame method .selectExpr()takes SQL expressions as a string:

flights.selectExpr("air_time/60 as duration_hrs")

with the SQL as keyword being equivalent to the .alias() method. To select multiple columns, you can pass multiple strings.

# Define avg_speed
avg_speed = (flights.distance/(flights.air_time/60)).alias("avg_speed")

# Select the correct columns
speed1 = flights.select("origin", "dest", "tailnum", avg_speed)

# Create the same table using a SQL expression
speed2 = flights.selectExpr("origin", "dest", "tailnum", "distance/(air_time/60) as avg_speed")

Groupby

df.groupBy().min("col").show()

This creates a GroupedData object (so you can use the .min() method), then finds the minimum value in col, and returns it as a DataFrame.

# Find the shortest flight from PDX in terms of distance
flights.filter(flights.origin == "PDX").groupBy().min("distance").show()

# Find the longest flight from SEA in terms of air time
flights.filter(flights.origin == "SEA").groupBy().max("air_time").show()

# Average duration of Delta flights
flights.filter(flights.carrier == "DL").filter(flights.origin == "SEA").groupBy().avg("air_time").show()

# Total hours in the air
flights.withColumn("duration_hrs", flights.air_time/60).groupBy().sum("duration_hrs").show()

# Group by tailnum
by_plane = flights.groupBy("tailnum")

# Number of flights each plane made
by_plane.count().show()

# Group by origin
by_origin = flights.groupBy("origin")

# Average duration of flights from PDX and SEA
by_origin.avg("air_time").show()

还可以在aggregate functions from the pyspark.sql.functions叠加std之类的。

# Import pyspark.sql.functions as F
import pyspark.sql.functions as F

# Group by month and dest
by_month_dest = flights.groupBy("month", "dest")

# Average departure delay by month and destination
by_month_dest.avg("dep_delay").show()

# Standard deviation of departure delay
by_month_dest.agg(F.stddev("dep_delay")).show()

Join

# Rename the faa column
airports = airports.withColumnRenamed("faa", "dest")

# Join the DataFrames
flights_with_airports = flights.join(airports, on="dest", how="leftouter")

Cast

Spark接收的是numerics，所以要么double要么integer。To convert the type of a column using the .cast()method, you can write code like this:

dataframe = dataframe.withColumn("col", dataframe.col.cast("new_type"))

比如，make a boolean:

# Create is_late
model_data = model_data.withColumn("is_late", model_data.arr_delay > 0)

# Convert to an integer
model_data = model_data.withColumn("label", model_data.is_late.cast("integer"))

Onehot Encoder, StringIndexer

# Create a StringIndexer
carr_indexer = StringIndexer(inputCol="carrier", outputCol="carrier_index")

# Create a OneHotEncoder
carr_encoder = OneHotEncoder(inputCol="carrier_index", outputCol="carrier_fact")

VectorAssembler

# Make a VectorAssembler
vec_assembler = VectorAssembler(inputCols=["month", "air_time", "carrier_fact", "dest_fact", "plane_age"], outputCol="features")

Pipeline

# Import Pipeline
from pyspark.ml import Pipeline

# Make the pipeline
flights_pipe = Pipeline(stages=[dest_indexer, dest_encoder, carr_indexer, carr_encoder, vec_assembler])

# Fit and transform the data
piped_data = flights_pipe.fit(model_data).transform(model_data)

Split the data

# Split the data into training and test sets
training, test = piped_data.randomSplit([.6, .4])

cross validation

# Import the evaluation submodule
import pyspark.ml.evaluation as evals

# Create a BinaryClassificationEvaluator
evaluator = evals.BinaryClassificationEvaluator(metricName="areaUnderROC")

Tuning

# Import the tuning submodule
import pyspark.ml.tuning as tune

# Create the parameter grid
grid = tune.ParamGridBuilder()

# Add the hyperparameter
grid = grid.addGrid(lr.regParam, np.arange(0, .1, .01))
grid = grid.addGrid(lr.elasticNetParam, [0, 1])

# Build the grid
grid = grid.build()

# Create the CrossValidator
cv = tune.CrossValidator(estimator=lr,
                         estimatorParamMaps=grid,
                         evaluator=evaluator
                         )
                         
# Fit cross validation models
models = cv.fit(training)

# Extract the best model
best_lr = models.bestModel

PreviousChapter7 Big Data and Spark NextData Cleaning with Apache Spark

Last updated 4 years ago

# Import SparkSession from pyspark.sql from pyspark.sql import SparkSession # Create my_spark my_spark = SparkSession.builder.getOrCreate() # Print the tables in the catalog print(spark.catalog.listTables())

# Create pd_temp pd_temp = pd.DataFrame(np.random.random(10)) # Create spark_temp from pd_temp spark_temp = spark.createDataFrame(pd_temp) # Examine the tables in the catalog print(spark.catalog.listTables()) # Add spark_temp to the catalog spark_temp.createOrReplaceTempView("temp") # Examine the tables in the catalog again print(spark.catalog.listTables())

# Select the first set of columns selected1 = flights.select("tailnum", "origin", "dest") # Select the second set of columns temp = flights.select(flights.origin, flights.dest, flights.carrier) # Define first filter filterA = flights.origin == "SEA" # Define second filter filterB = flights.dest == "PDX" # Filter the data, first by filterA then by filterB selected2 = temp.filter(filterA).filter(filterB)

# Define avg_speed avg_speed = (flights.distance/(flights.air_time/60)).alias("avg_speed") # Select the correct columns speed1 = flights.select("origin", "dest", "tailnum", avg_speed) # Create the same table using a SQL expression speed2 = flights.selectExpr("origin", "dest", "tailnum", "distance/(air_time/60) as avg_speed")

# Find the shortest flight from PDX in terms of distance flights.filter(flights.origin == "PDX").groupBy().min("distance").show() # Find the longest flight from SEA in terms of air time flights.filter(flights.origin == "SEA").groupBy().max("air_time").show() # Average duration of Delta flights flights.filter(flights.carrier == "DL").filter(flights.origin == "SEA").groupBy().avg("air_time").show() # Total hours in the air flights.withColumn("duration_hrs", flights.air_time/60).groupBy().sum("duration_hrs").show() # Group by tailnum by_plane = flights.groupBy("tailnum") # Number of flights each plane made by_plane.count().show() # Group by origin by_origin = flights.groupBy("origin") # Average duration of flights from PDX and SEA by_origin.avg("air_time").show()

# Import pyspark.sql.functions as F import pyspark.sql.functions as F # Group by month and dest by_month_dest = flights.groupBy("month", "dest") # Average departure delay by month and destination by_month_dest.avg("dep_delay").show() # Standard deviation of departure delay by_month_dest.agg(F.stddev("dep_delay")).show()

# Create a StringIndexer carr_indexer = StringIndexer(inputCol="carrier", outputCol="carrier_index") # Create a OneHotEncoder carr_encoder = OneHotEncoder(inputCol="carrier_index", outputCol="carrier_fact")

# Import Pipeline from pyspark.ml import Pipeline # Make the pipeline flights_pipe = Pipeline(stages=[dest_indexer, dest_encoder, carr_indexer, carr_encoder, vec_assembler]) # Fit and transform the data piped_data = flights_pipe.fit(model_data).transform(model_data)

# Import the tuning submodule import pyspark.ml.tuning as tune # Create the parameter grid grid = tune.ParamGridBuilder() # Add the hyperparameter grid = grid.addGrid(lr.regParam, np.arange(0, .1, .01)) grid = grid.addGrid(lr.elasticNetParam, [0, 1]) # Build the grid grid = grid.build() # Create the CrossValidator cv = tune.CrossValidator(estimator=lr, estimatorParamMaps=grid, evaluator=evaluator ) # Fit cross validation models models = cv.fit(training) # Extract the best model best_lr = models.bestModel