Sequence to Dataframe


val myManualSchema = new StructType(Array(
  new StructField("some", StringType, true),
  new StructField("col", StringType, true),
  new StructField("names", LongType, false)))

val myRows = Seq(Row("Hello", null, 1L))
val myRDD = spark.sparkContext.parallelize(myRows)
val myDf = spark.createDataFrame(myRDD, myManualSchema)


      

Select

df.select("DEST_COUNTRY_NAME").show(2)


      

Select and rename

df.select(col("DEST_COUNTRY_NAME").alias("dest")).show(2)


      

Filter

df.where(col("count") < 2)
  .where(col("ORIGIN_COUNTRY_NAME") == "Croatia")
  .show(2)


      

Bind Filter Select

val DOTCodeFilter = col("StockCode") === "DOT"
val priceFilter = col("UnitPrice") > 600
val descripFilter = col("Description").contains("POSTAGE") 
df.withColumn("isExpensive", 
              DOTCodeFilter.and(priceFilter.or(descripFilter)))
  .where("isExpensive")
  .select("unitPrice", "isExpensive")
  .show(5)  


      

Complex Types

df.select(split(col("Description"), " ").alias("array_col"))
  .selectExpr("array_col[0]").show(2 false)


      

User Defined Functions (UDFs)

def mul3(double_value):
   return double_value * 3

val mul3udf = udf(mul3(_:Double):Double)

df.select(mul3udf(col("num"))).show()


      

Join (inner default,and outer)


val joinExpression = person.col("graduate_program") 
                     === graduateProgram.col("id")
person.join(graduateProgram, joinExpression).show()


val joinExpression = person.col("graduate_program") 
                     === graduateProgram.col("id")
person.join(graduateProgram, joinExpression, "outer").show()


      

Aggregate on Dataframe (agg and select)

df.select(
     count("Quantity").alias("ttransactions"),
     sum("Quantity").alias("tpurchases"),
     avg("Quantity").alias("avg_purchases"),
     expr("mean(Quantity)").alias("mean_purchases"))
   .selectExpr(
     "tpurchases/ttransactions",
     "avg_purchases",
     "mean_purchases").show()


      

Aggregate on groups

df.groupBy("InvoiceNo", "CustomerId").count().show()

df.groupBy("InvoiceNo").agg(
   count("Quantity").alias("quan"),
   sum("Quantity")).show()


      

Window Spec

val dfWithDate = df.withColumn("date", to_date(col("InvoiceDate"),
                                               "MM/d/yyyy H:mm"))

dfWithDate.createOrReplaceTempView("dfWithDate")

val windowSpec = Window
                 .partitionBy("CustomerId", "date")
                 .orderBy(col("Quantity").desc)
                 .rowsBetween(Window.unboundedPreceding, 
                              Window.currentRow)

val maxPurchaseQuantity = max(col("Quantity")).over(windowSpec)
val purchaseDenseRank = dense_rank().over(windowSpec)
val purchaseRank = rank().over(windowSpec)

dfWithDate.where("CustomerId IS NOT NULL").orderBy("CustomerId")
 .select(
   col("CustomerId"),col("date"),
   col("Quantity"),
   purchaseRank.alias("rank"),
   purchaseDenseRank.alias("denseRank"),
   maxPurchaseQuantity.alias("maxq")).show()


      

Datasets create and Filter


case class Flight(DEST_COUNTRY_NAME: String,
                  ORIGIN_COUNTRY_NAME: String,
                  count: BigInt)
val flightsDF = spark.read
               .parquet("/data/flight-data/parquet/2010-summary.parquet/")

val flights = flightsDF.as[Flight]

def originIsDestination(flight_row: Flight): Boolean = {
  return flight_row.ORIGIN_COUNTRY_NAME 
         ==
         flight_row.DEST_COUNTRY_NAME
}

flights.filter(flight_row => originIsDestination(flight_row))


      

WordCount

val counts = textFile
             .flatMap(line => line.split(" "))
             .map(word => (word, 1))
             .reduceByKey(_ + _)


      

Streaming

val readStream = spark.readStream.schema(dataSchema)
                .option("maxFilesPerTrigger", 1)
                .json("/data/activity-data")

streaming.withColumn("stairs", expr("gt like '%stairs%'"))
         .where("stairs")
         .where("gt is not null")
         .select("gt", "model", "arrival_time", "creation_time")
         .writeStream
         .queryName("simple_transform")
         .format("memory")
         .outputMode("append")
         .start()