【Apache Spark】Apache Spark API By Example

1. The ``where()`` clause is equivalent to ``filter()``.
   
2. 
   
3. Copy to clipboardCopy
   
4. val whereDF = explodeDF.where(($"firstName" === "xiangrui") || ($"firstName" === "michael")).sort($"lastName".asc)
   
5. display(whereDF)

复制代码

1. Replace ``null`` values with ``–`` using DataFrame Na functions.
   
2. 
   
3. Copy to clipboardCopy
   
4. val naFunctions = explodeDF.na
   
5. val nonNullDF = naFunctions.fill("--")
   
6. display(nonNullDF)

复制代码

1. Retrieve only rows with missing firstName or lastName.
   
2. 
   
3. Copy to clipboardCopy
   
4. val filterNonNullDF = nonNullDF.filter($"firstName" === "" || $"lastName" === "").sort($"email".asc)
   
5. display(filterNonNullDF)

复制代码

1. Example aggregations using ``agg()`` and ``countDistinct()``.
   
2. 
   
3. Copy to clipboardCopy
   
4. import org.apache.spark.sql.functions._
   
5. Copy to clipboardCopy
   
6. // Find the distinct (firstName, lastName) combinations
   
7. val countDistinctDF = nonNullDF.select($"firstName", $"lastName")
   
8.   .groupBy($"firstName", $"lastName")
   
9.   .agg(countDistinct($"firstName") as "distinct_first_names")
   
10. display(countDistinctDF)

复制代码

1. Compare the DataFrame and SQL Query Physical Plans (Hint: They should be the same.)
   
2. 
   
3. Copy to clipboardCopy
   
4. countDistinctDF.explain()
   
5. Copy to clipboardCopy
   
6. // register the DataFrame as a temp table so that we can query it using SQL
   
7. nonNullDF.registerTempTable("databricks_df_example")
   
8. 
   
9. // Perform the same query as the DataFrame above and return ``explain``
   
10. sqlContext.sql("""
    
11. SELECT firstName, lastName, count(distinct firstName) as distinct_first_names
    
12. FROM databricks_df_example
    
13. GROUP BY firstName, lastName
    
14. """).explain
    
15. Copy to clipboardCopy
    
16. // Sum up all the salaries
    
17. val salarySumDF = nonNullDF.agg("salary" -> "sum")
    
18. display(salarySumDF)

复制代码

1. Creating Datasets
   
2. 
   
3. You can simply call .toDS() on a sequence to convert the sequence to a Dataset.
   
4. 
   
5. Copy to clipboardCopy
   
6. val dataset = Seq(1, 2, 3).toDS()
   
7. dataset.show()
   
8. If you have a sequence of case classes, calling .toDS() will provide a dataset with all the necessary fields in the dataset.
   
9. 
   
10. Copy to clipboardCopy
    
11. case class Person(name: String, age: Int)
    
12. 
    
13. val personDS = Seq(Person("Max", 33), Person("Adam", 32), Person("Muller", 62)).toDS()
    
14. personDS.show()
    
15. Creating Datasets from a RDD You can call rdd.toDS() to convert an RDD into a Dataset.
    
16. 
    
17. Copy to clipboardCopy
    
18. val rdd = sc.parallelize(Seq((1, "Spark"), (2, "Databricks")))
    
19. val integerDS = rdd.toDS()
    
20. integerDS.show()
    
21. Creating Datasets from a DataFrame You can call df.as[SomeCaseClass] to convert the DataFrame to a Dataset.
    
22. 
    
23. Copy to clipboardCopy
    
24. case class Company(name: String, foundingYear: Int, numEmployees: Int)
    
25. val inputSeq = Seq(Company("ABC", 1998, 310), Company("XYZ", 1983, 904), Company("NOP", 2005, 83))
    
26. val df = sc.parallelize(inputSeq).toDF()
    
27. 
    
28. val companyDS = df.as[Company]
    
29. companyDS.show()
    
30. You can also deal with tuples while converting a DataFrame to Dataset without using a case class
    
31. 
    
32. Copy to clipboardCopy
    
33. val rdd = sc.parallelize(Seq((1, "Spark"), (2, "Databricks"), (3, "Notebook")))
    
34. val df = rdd.toDF("Id", "Name")
    
35. 
    
36. val dataset = df.as[(Int, String)]
    
37. dataset.show()

复制代码

1. Word Count Example
   
2. 
   
3. Copy to clipboardCopy
   
4. val wordsDataset = sc.parallelize(Seq("Spark I am your father", "May the spark be with you", "Spark I am your father")).toDS()
   
5. val groupedDataset = wordsDataset.flatMap(_.toLowerCase.split(" "))
   
6.                                  .filter(_ != "")
   
7.                                  .groupBy("value")
   
8. val countsDataset = groupedDataset.count()
   
9. countsDataset.show()

复制代码

1. Converting Datasets to DataFrames
   
2. The above 2 examples dealt with using pure Datasets APIs. You can also easily move from Datasets to DataFrames and leverage the DataFrames APIs. The below example shows the word count example that uses both Datasets and DataFrames APIs.
   
3. 
   
4. Copy to clipboardCopy
   
5. import org.apache.spark.sql.functions._
   
6. 
   
7. val wordsDataset = sc.parallelize(Seq("Spark I am your father", "May the spark be with you", "Spark I am your father")).toDS()
   
8. val result = wordsDataset
   
9.                .flatMap(_.split(" "))               // Split on whitespace
   
10.                .filter(_ != "")                     // Filter empty words
    
11.                .map(_.toLowerCase())
    
12.                .toDF()                              // Convert to DataFrame to perform aggregation / sorting
    
13.                .groupBy($"value")                   // Count number of occurences of each word
    
14.                .agg(count("*") as "numOccurances")
    
15.                .orderBy($"numOccurances" desc)      // Show most common words first
    
16. result.show()

复制代码

谢谢分享