【Apache Spark】Apache Spark API By Example

相似文件 换一批

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

1. Apache Spark API By Example
   
2. A Command Reference for Beginners
   
3. Matthias Langer, Zhen He
   
4. Department of Computer Science and Computer Engineering
   
5. La Trobe University
   
6. Bundoora, VIC 3086
   
7. Australia
   
8. m.langer@latrobe.edu.au, z.he@latrobe.edu.au
   
9. May 31, 2014

复制代码

本帖隐藏的内容

Apache Spark API By Example.rar (258.94 KB) 本附件包括：

• Apache Spark API By Example.pdf

2017-3-8 10:25:17 上传

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：Apache Spark example apache Spark ExamP 2014

1. Creating DataFrames with Python
   
2. 
   
3. Copy to clipboardCopy
   
4. # import pyspark class Row from module sql
   
5. from pyspark.sql import *
   
6. 
   
7. # Create Example Data - Departments and Employees
   
8. 
   
9. # Create the Departments
   
10. department1 = Row(id='123456', name='Computer Science')
    
11. department2 = Row(id='789012', name='Mechanical Engineering')
    
12. department3 = Row(id='345678', name='Theater and Drama')
    
13. department4 = Row(id='901234', name='Indoor Recreation')
    
14. 
    
15. # Create the Employees
    
16. Employee = Row("firstName", "lastName", "email", "salary")
    
17. employee1 = Employee('michael', 'armbrust', 'no-reply@berkeley.edu', 100000)
    
18. employee2 = Employee('xiangrui', 'meng', 'no-reply@stanford.edu', 120000)
    
19. employee3 = Employee('matei', None, 'no-reply@waterloo.edu', 140000)
    
20. employee4 = Employee(None, 'wendell', 'no-reply@berkeley.edu', 160000)
    
21. 
    
22. # Create the DepartmentWithEmployees instances from Departments and Employees
    
23. departmentWithEmployees1 = Row(department=department1, employees=[employee1, employee2])
    
24. departmentWithEmployees2 = Row(department=department2, employees=[employee3, employee4])
    
25. departmentWithEmployees3 = Row(department=department3, employees=[employee1, employee4])
    
26. departmentWithEmployees4 = Row(department=department4, employees=[employee2, employee3])
    
27. 
    
28. print department1
    
29. print employee2
    
30. print departmentWithEmployees1.employees[0].email
    
31. Create the first DataFrame from a list of the rows.
    
32. 
    
33. Copy to clipboardCopy
    
34. departmentsWithEmployeesSeq1 = [departmentWithEmployees1, departmentWithEmployees2]
    
35. df1 = sqlContext.createDataFrame(departmentsWithEmployeesSeq1)
    
36. 
    
37. display(df1)
    
38. Create a second DataFrame from a list of rows.
    
39. 
    
40. Copy to clipboardCopy
    
41. departmentsWithEmployeesSeq2 = [departmentWithEmployees3, departmentWithEmployees4]
    
42. df2 = sqlContext.createDataFrame(departmentsWithEmployeesSeq2)
    
43. 
    
44. display(df2)

复制代码

1. An example using Pandas & Matplotlib Integration
   
2. Copy to clipboardCopy
   
3. import pandas as pd
   
4. import matplotlib.pyplot as plt
   
5. plt.clf()
   
6. pdDF = nonNullDF.toPandas()
   
7. pdDF.plot(x='firstName', y='salary', kind='bar', rot=45)
   
8. display()
   
9. Cleanup: Remove the parquet file.
   
10. 
    
11. Copy to clipboardCopy
    
12. dbutils.fs.rm("/tmp/databricks-df-example.parquet", True)

复制代码

1. Create DataFrames using Scala
   
2. Copy to clipboardCopy
   
3. // Create the case classes for our domain
   
4. case class Department(id: String, name: String)
   
5. case class Employee(firstName: String, lastName: String, email: String, salary: Int)
   
6. case class DepartmentWithEmployees(department: Department, employees: Seq[Employee])
   
7. 
   
8. // Create the Departments
   
9. val department1 = new Department("123456", "Computer Science")
   
10. val department2 = new Department("789012", "Mechanical Engineering")
    
11. val department3 = new Department("345678", "Theater and Drama")
    
12. val department4 = new Department("901234", "Indoor Recreation")
    
13. 
    
14. // Create the Employees
    
15. val employee1 = new Employee("michael", "armbrust", "no-reply@berkeley.edu", 100000)
    
16. val employee2 = new Employee("xiangrui", "meng", "no-reply@stanford.edu", 120000)
    
17. val employee3 = new Employee("matei", null, "no-reply@waterloo.edu", 140000)
    
18. val employee4 = new Employee(null, "wendell", "no-reply@princeton.edu", 160000)
    
19. 
    
20. // Create the DepartmentWithEmployees instances from Departments and Employees
    
21. val departmentWithEmployees1 = new DepartmentWithEmployees(department1, Seq(employee1, employee2))
    
22. val departmentWithEmployees2 = new DepartmentWithEmployees(department2, Seq(employee3, employee4))
    
23. val departmentWithEmployees3 = new DepartmentWithEmployees(department3, Seq(employee1, employee4))
    
24. val departmentWithEmployees4 = new DepartmentWithEmployees(department4, Seq(employee2, employee3))
    
25. Create the first DataFrame from a List of the Case Classes.
    
26. 
    
27. Copy to clipboardCopy
    
28. val departmentsWithEmployeesSeq1 = Seq(departmentWithEmployees1, departmentWithEmployees2)
    
29. val df1 = departmentsWithEmployeesSeq1.toDF()
    
30. display(df1)
    
31. Create a 2nd DataFrame from a List of Case Classes.
    
32. 
    
33. Copy to clipboardCopy
    
34. val departmentsWithEmployeesSeq2 = Seq(departmentWithEmployees3, departmentWithEmployees4)
    
35. val df2 = departmentsWithEmployeesSeq2.toDF()
    
36. display(df2)

复制代码

1. Flattening using Scala
   
2. If your data has several levels of nesting, here is a helper function to flatten your DataFrame to make it easier to work with.
   
3. 
   
4. Copy to clipboardCopy
   
5. val veryNestedDF = Seq(("1", (2, (3, 4)))).toDF()
   
6. Copy to clipboardCopy
   
7. import org.apache.spark.sql._
   
8. import org.apache.spark.sql.functions._
   
9. import org.apache.spark.sql.types._
   
10. 
    
11. implicit class DataFrameFlattener(df: DataFrame) {
    
12.   def flattenSchema: DataFrame = {
    
13.     df.select(flatten(Nil, df.schema): _*)
    
14.   }
    
15. 
    
16.   protected def flatten(path: Seq[String], schema: DataType): Seq[Column] = schema match {
    
17.     case s: StructType => s.fields.flatMap(f => flatten(path :+ f.name, f.dataType))
    
18.     case other => col(path.map(n => s"`$n`").mkString(".")).as(path.mkString(".")) :: Nil
    
19.   }
    
20. }
    
21. Copy to clipboardCopy
    
22. display(veryNestedDF)
    
23. Copy to clipboardCopy
    
24. display(veryNestedDF.flattenSchema)
    
25. Cleanup: Remove the parquet file.
    
26. 
    
27. Copy to clipboardCopy
    
28. dbutils.fs.rm("/tmp/databricks-df-example.parquet", true)

复制代码

1. Union 2 DataFrames.
   
2. 
   
3. Copy to clipboardCopy
   
4. val unionDF = df1.unionAll(df2)
   
5. display(unionDF)

复制代码

1. Write the Unioned DataFrame to a Parquet file.
   
2. 
   
3. Copy to clipboardCopy
   
4. // Remove the file if it exists
   
5. dbutils.fs.rm("/tmp/databricks-df-example.parquet", true)
   
6. unionDF.write.parquet("/tmp/databricks-df-example.parquet")

复制代码

1. Read a DataFrame from the Parquet file.
   
2. 
   
3. Copy to clipboardCopy
   
4. val parquetDF = sqlContext.read.parquet("/tmp/databricks-df-example.parquet")
   
5. Copy to clipboardCopy
   
6. val explodeDF = parquetDF.explode($"employees") {
   
7.   case Row(employee: Seq[Row]) => employee.map{ employee =>
   
8.     val firstName = employee(0).asInstanceOf[String]
   
9.     val lastName = employee(1).asInstanceOf[String]
   
10.     val email = employee(2).asInstanceOf[String]
    
11.     val salary = employee(3).asInstanceOf[Int]
    
12.     Employee(firstName, lastName, email, salary)
    
13.   }
    
14. }.cache()
    
15. display(explodeDF)
    
16. Copy to clipboardCopy
    
17. explodeDF

复制代码

1. Use ``filter()`` to return only the rows that match the given predicate.
   
2. 
   
3. Copy to clipboardCopy
   
4. val filterDF = explodeDF
   
5.   .filter($"firstName" === "xiangrui" || $"firstName" === "michael")
   
6.   .sort($"lastName".asc)
   
7. display(filterDF)

复制代码