Apache Spark数据分析教程（二）：Spark SQL

dataframe.filter(dataframe.author =="Geoffrey Chandler").count() 102

Michael, 29 Andy, 30 Justin, 19

# Import data types from pyspark.sql.types import * # Create a RDD from `people.txt` # then convert each line to a tuple. lines = sc.textFile("examples/src/main/resources/people.txt") parts = lines.map(lambda l: l.split(",")) people = parts.map(lambda p: (p[0], p[1].strip())) # encode the schema in a string. schemaString = "name age" # Create a type fields fields = [StructField(field_name, StringType(), True) /  for field_name in schemaString.split()] # Create the schema schema = StructType(fields) # Apply the schema to the RDD. schemaPeople = sqlContext.createDataFrame(people, schema) # In order to query data you need # to register the DataFrame as a table. schemaPeople.registerTempTable("people") # Using sql query all the name from the table results = sqlContext.sql("SELECT name FROM people") # The results of SQL queries are RDDs # and support all the normal RDD operations. names = results.map(lambda p: "Name: " + p.name) for name in names.collect():   print name 

Name: Michael Name: Andy Name: Justin

# First we need to import the following Row class from pyspark.sql import SQLContext, Row # Create a RDD peopleAge, # when this is done the RDD will # be partitioned into three partitions peopleAge = sc.textFile("examples/src/main/resources/people.txt") # Since name and age are separated by a comma let's split them parts = peopleAge.map(lambda l: l.split(",")) # Every line in the file will represent a row # with 2 columns name and age. # After this line will have a table called people people = parts.map(lambda p: Row(name=p[0], age=int(p[1]))) # Using the RDD create a DataFrame schemaPeople = sqlContext.createDataFrame(people) # In order to do sql query on a dataframe, # you need to register it as a table schemaPeople.registerTempTable("people") # Finally we are ready to use the DataFrame. # Let's query the adults that are aged between 21 and 50 adults = sqlContext.sql("SELECT name FROM people /        WHERE age >= 21 AND age <= 50") # loop through names and ages adults = adults.map(lambda p: "Name: " + p.name) for Adult in adults.collect():   print Adult

Name: Michael Name: Andy

dependencies {     //Spark and Cassandra connector to work with java    compile 'com.datastax.spark:spark-cassandra-connector_2.10:1.1.1-rc4'     compile 'com.datastax.spark:spark-cassandra-connector-java_2.10:1.1.1'     compile 'org.apache.spark:spark-streaming_2.10:1.5.0' }

private void createSchema(JavaSparkContext sc) {     CassandraConnector connector =        CassandraConnector.apply(sc.getConf());     try (Session session = connector.openSession()) {      session.execute(deletekeyspace);      session.execute(keyspace);      session.execute("USE todolist");      session.execute(table);      session.execute(tableRDD);     } } 

/* Delete keyspace todolist if exists. */ String deletekeyspace = "DROP KEYSPACE IF EXISTS todolist"; /* Create keyspace todolist. */ String keyspace = "CREATE KEYSPACE IF NOT EXISTS todolist" +   " WITH replication = {'class': 'SimpleStrategy'," +   " 'replication_factor':1}"; /* Create table todolisttable. */ String table = "CREATE TABLE todolist.todolisttable(" +  + " id text PRIMARY KEY, "  + " description text, "  + " category text, "  + " date timestamp )"; /* Create table temp. */ String tableRDD = "CREATE TABLE todolist.temp(id text PRIMARY KEY, "  + "description text, "  + "category text )"; 

TodoItem item = new TodoItem("George", "Buy a new computer", "Shopping");    TodoItem item2 = new TodoItem("John", "Go to the gym", "Sport");     TodoItem item3 = new TodoItem("Ron", "Finish the homework", "Education");     TodoItem item4 = new TodoItem("Sam", "buy a car", "Shopping");     TodoItem item5 = new TodoItem("Janet", "buy groceries", "Shopping");     TodoItem item6 = new TodoItem("Andy", "go to the beach", "Fun");     TodoItem item7 = new TodoItem("Paul", "Prepare lunch", "Coking"); //index data     String task1 = "INSERT INTO todolisttable (ID, Description, Category, Date)"      + item.toString();     String task2 = "INSERT INTO todolisttable (ID, Description, Category, Date)"      + item2.toString();     String task3 = "INSERT INTO todolisttable (ID, Description, Category, Date)"      + item3.toString();     String task4 = "INSERT INTO todolisttable (ID, Description, Category, Date)"      + item4.toString();     String task5 = "INSERT INTO todolisttable (ID, Description, Category, Date)"      + item5.toString();     String task6 = "INSERT INTO todolisttable (ID, Description, Category, Date)"      + item6.toString();     String task7 = "INSERT INTO todolisttable (ID, Description, Category, Date)"      + item7.toString(); 

private void queryData(JavaSparkContext sc) {         CassandraConnector connector =                     CassandraConnector.apply(sc.getConf());         try (Session session = connector.openSession()) {             ResultSet results = session.execute(query);             System.out.println("Query all results from cassandra:/n" + results.all());         }

public  void accessTableWitRDD(JavaSparkContext sc){  JavaRDD<String> cassandraRDD = javaFunctions(sc).cassandraTable("todolist", "todolisttable")    .map(new Function<CassandraRow, String>() {     @Override     public String call(CassandraRow cassandraRow) throws Exception {      return cassandraRow.toString();     }    }); } 

System.out.println("/nData as CassandraRows from a RDD: /n" + StringUtils.join(cassandraRDD.toArray(), "/n"));

public void saveRDDToCass(JavaSparkContext sc) {         List<TodoItem> todos = Arrays.asList(         new TodoItem("George", "Buy a new computer", "Shopping"),         new TodoItem("John", "Go to the gym", "Sport"),         new TodoItem("Ron", "Finish the homework", "Education"),         new TodoItem("Sam", "buy a car", "Shopping"),         new TodoItem("Janet", "buy groceries", "Shopping"),         new TodoItem("Andy", "go to the beach", "Fun"),         new TodoItem("Paul", "Prepare lunch", "Coking")         );         JavaRDD<TodoItem> rdd = sc.parallelize(todos);         javaFunctions(rdd).writerBuilder("todolist", "temp", mapToRow(TodoItem.class)).saveToCassandra();

String query1 = "SELECT * FROM todolist.temp"; ResultSet results1 = session.execute(query1); System.out.println("/nQuery all results from temp" +  " table after saving a RDD into Cassandra:/n" +  results1.all());

dependencies {  compile 'org.apache.spark:spark-sql_2.10:1.5.0' }

为Cassandra 创建SparkSpark configuration对象

SparkConf conf = new SparkConf(); conf.setAppName("TODO sparkSQL and cassandra"); conf.setMaster("local"); conf.set("spark.cassandra.connection.host", "localhost");

创建Spark Context对象 (JavaSparkContext)。

创建Spark Context

JavaSparkContextsc=newJavaSparkContext(conf);

创建SQLContext对象以便使用SQL连接Cassandra：

创建Spark SQL Context

JavaSparkContext sc = new JavaSparkContext(conf);

通过SQLContext便能注册RDD并利用Spark SQL进行查询操作。

然后，创建RDD对象（rdd）并加载数据(TodoItems)：

RDD 加载 TodoItems

List<TodoItem> todos = Arrays.asList(  new TodoItem("George", "Buy a new computer", "Shopping"),  new TodoItem("John", "Go to the gym", "Sport"),  new TodoItem("Ron", "Finish the homework", "Education"),  new TodoItem("Sam", "buy a car", "Shopping"),  new TodoItem("Janet", "buy groceries", "Shopping"),  new TodoItem("Andy", "go to the beach", "Fun"),  new TodoItem("Paul", "Prepare lunch", "Cooking") ); JavaRDD<TodoItem> rdd = sc.parallelize(todos); 

需要注意的是我们使用parallelize方法将所有的Todo数据加载到整个Spark集群。 JavaRDD 通过context.parallelize方法产生。

然后，通过sqlContext创建DataFrame：

通过sqlContext创建DataFrame

DataFrame dataframe = sqlContext.createDataFrame(rdd, TodoItem.class);

Dataframe从 TodoItem.class获取对应的schema。

然后，将dataframe注册成名为todo的表：

将DataFrame注册成名为todo的表

sqlContext.registerDataFrameAsTable(dataframe, "todo");

这样后面便可以使用todo进行DataFrame数据的查询。

到这一步便可以使用Spark SQL提供的所有操作，首先对todo items进行计数，它将加载数据到内存以便进行更快速的查询操作：

获取DataFrame中TODO items的数量

System.out.println("Total number of TodoItems = [" + rdd.count() + "]/n");

最后，使用SQL进行数据的查询：

使用Spark SQL查询Todo Items并显示查询结果

DataFrame result = sqlContext.sql("SELECT * from todo");         System.out.println("Show the DataFrame result:/n");         result.show();         System.out.println("Select the id column and show its contents:/n");         result.select("id").show();

可以在本文最后的运行部分获取相关代码及如何运行这些代码的教程。这里给出本文例子的完整代码。

完整代码清单

SparkApp.java清单

package com.example; import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Session; import com.datastax.spark.connector.cql.CassandraConnector; import com.datastax.spark.connector.japi.CassandraRow; import org.apache.commons.lang.StringUtils; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.Function; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.Serializable; import java.util.Arrays; import java.util.List; import static com.datastax.spark.connector.japi.CassandraJavaUtil.javaFunctions; import static com.datastax.spark.connector.japi.CassandraJavaUtil.mapToRow; //import org.apache.cassandra.cql.BatchStatement; /**  * Created by fadi on 5/18/15.  */ public class SparkApp implements Serializable {  static final Logger logger = LoggerFactory.getLogger(SparkApp.class);  TodoItem item = new TodoItem("George", "Buy a new computer", "Shopping");  TodoItem item2 = new TodoItem("John", "Go to the gym", "Sport");  TodoItem item3 = new TodoItem("Ron", "Finish the homework", "Education");  TodoItem item4 = new TodoItem("Sam", "buy a car", "Shopping");  TodoItem item5 = new TodoItem("Janet", "buy groceries", "Shopping");  TodoItem item6 = new TodoItem("Andy", "go to the beach", "Fun");  TodoItem item7 = new TodoItem("Paul", "Prepare lunch", "Coking");  String keyspace = "CREATE KEYSPACE IF NOT EXISTS todolist  WITH replication = {'class': 'SimpleStrategy', 'replication_factor':1}";  //index data  String task1 = "INSERT INTO todolisttable (ID, Description, Category, Date)"    + item.toString();  String task2 = "INSERT INTO todolisttable (ID, Description, Category, Date)"    + item2.toString();  String task3 = "INSERT INTO todolisttable (ID, Description, Category, Date)"    + item3.toString();  String task4 = "INSERT INTO todolisttable (ID, Description, Category, Date)"    + item4.toString();  String task5 = "INSERT INTO todolisttable (ID, Description, Category, Date)"    + item5.toString();  String task6 = "INSERT INTO todolisttable (ID, Description, Category, Date)"    + item6.toString();  String task7 = "INSERT INTO todolisttable (ID, Description, Category, Date)"    + item7.toString();  //delete keyspace  String deletekeyspace = "DROP KEYSPACE IF EXISTS todolist";  //delete table  String deletetable = "DROP TABLE todolisttable";  //create table  String table = "CREATE TABLE todolist.todolisttable(id text PRIMARY KEY, "    + "description text, "    + "category text, "    + "date timestamp )";  String tableRDD = "CREATE TABLE todolist.temp(id text PRIMARY KEY, "    + "description text, "    + "category text )";  //Query all data  String query = "SELECT * FROM todolist.todolisttable";  String query1 = "SELECT * FROM todolist.temp";  //Update table  String update = "UPDATE todolisttable SET Category='Fun',Description='Go to the beach' WHERE ID='Ron'";  //Deleting data where the index id = George  String delete = "DELETE FROM todolisttable WHERE ID='George'";  //Deleting all data  String deleteall = "TRUNCATE todolisttable"; //---------------------------------------------------------------------------------  private transient SparkConf conf;  private SparkApp(SparkConf conf) {   this.conf = conf;  }  private void run() {   JavaSparkContext sc = new JavaSparkContext(conf);   createSchema(sc);   loadData(sc);   saveRDDToCassandra(sc);   queryData(sc);   accessTableWitRDD(sc);   sc.stop();  }  private void createSchema(JavaSparkContext sc) {   CassandraConnector connector = CassandraConnector.apply(sc.getConf());   try (Session session = connector.openSession()) {    session.execute(deletekeyspace);    session.execute(keyspace);    session.execute("USE todolist");    session.execute(table);    session.execute(tableRDD);   }  }  private void loadData(JavaSparkContext sc) {   CassandraConnector connector = CassandraConnector.apply(sc.getConf());   try (Session session = connector.openSession()) {    session.execute(task1);    session.execute(task2);    session.execute(task3);    session.execute(task4);    session.execute(task5);    session.execute(task6);    session.execute(task7);   }  }  private void queryData(JavaSparkContext sc) {   CassandraConnector connector = CassandraConnector.apply(sc.getConf());   try (Session session = connector.openSession()) {    ResultSet results = session.execute(query);    System.out.println("/nQuery all results from cassandra's todolisttable:/n" + results.all());    ResultSet results1 = session.execute(query1);    System.out.println("/nSaving RDD into a temp table in casssandra then query all results from cassandra:/n" + results1.all());   }  }  public  void accessTableWitRDD(JavaSparkContext sc){   JavaRDD<String> cassandraRDD = javaFunctions(sc).cassandraTable("todolist", "todolisttable")     .map(new Function<CassandraRow, String>() {      @Override      public String call(CassandraRow cassandraRow) throws Exception {       return cassandraRow.toString();      }     });   System.out.println("/nReading Data from todolisttable in Cassandra with a RDD: /n" + StringUtils.join(cassandraRDD.toArray(), "/n"));   // javaFunctions(cassandraRDD).writerBuilder("todolist", "todolisttable", mapToRow(String.class)).saveToCassandra();  }  public void saveRDDToCassandra(JavaSparkContext sc) {   List<TodoItem> todos = Arrays.asList(     new TodoItem("George", "Buy a new computer", "Shopping"),     new TodoItem("John", "Go to the gym", "Sport"),     new TodoItem("Ron", "Finish the homework", "Education"),     new TodoItem("Sam", "buy a car", "Shopping"),     new TodoItem("Janet", "buy groceries", "Shopping"),     new TodoItem("Andy", "go to the beach", "Fun"),     new TodoItem("Paul", "Prepare lunch", "Coking")   );   JavaRDD<TodoItem> rdd = sc.parallelize(todos);   javaFunctions(rdd).writerBuilder("todolist", "temp", mapToRow(TodoItem.class)).saveToCassandra();  } //----------------------------------------------------------------------------------------------------------------------------  public static void main( String args[] )  {   SparkConf conf = new SparkConf();   conf.setAppName("TODO spark and cassandra");   conf.setMaster("local");   conf.set("spark.cassandra.connection.host", "localhost");   SparkApp app = new SparkApp(conf);   app.run();  } } 

SparkSQLApp.java清单

package com.example; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.sql.DataFrame; import org.apache.spark.sql.SQLContext; import java.util.Arrays; import java.util.List; /**  * Created by fadi on 6/14/15.  */ public class SparkSQLApp {  private transient SparkConf conf;  private SparkSQLApp(SparkConf conf) {   this.conf = conf;  }  private void run() {   JavaSparkContext sc = new JavaSparkContext(conf);    SQLContext sqlContext = new SQLContext(sc);   createDataframe(sc, sqlContext);   querySQLData(sqlContext);   sc.stop();  }  public void createDataframe(JavaSparkContext sc, SQLContext sqlContext ) {   List<TodoItem> todos = Arrays.asList(     new TodoItem("George", "Buy a new computer", "Shopping"),     new TodoItem("John", "Go to the gym", "Sport"),     new TodoItem("Ron", "Finish the homework", "Education"),     new TodoItem("Sam", "buy a car", "Shopping"),     new TodoItem("Janet", "buy groceries", "Shopping"),     new TodoItem("Andy", "go to the beach", "Fun"),     new TodoItem("Paul", "Prepare lunch", "Cooking")   );   JavaRDD<TodoItem> rdd = sc.parallelize(todos);   DataFrame dataframe =   sqlContext.createDataFrame(rdd, TodoItem.class);   sqlContext.registerDataFrameAsTable(dataframe, "todo");   System.out.println("Total number of TodoItems = [" + rdd.count() + "]/n");  }  public void querySQLData(SQLContext sqlContext) {   DataFrame result = sqlContext.sql("SELECT * from todo");   System.out.println("Show the DataFrame result:/n");   result.show();   System.out.println("Select the id column and show its contents:/n");   result.select("id").show();  }  public static void main( String args[] )  {   SparkConf conf = new SparkConf();   conf.setAppName("TODO sparkSQL and cassandra");   conf.setMaster("local");   conf.set("spark.cassandra.connection.host", "localhost");   SparkSQLApp app = new SparkSQLApp(conf);   app.run();  } } 

Todoitem.java清单

package com.example; import java.io.Serializable; import java.time.LocalDateTime; public class TodoItem implements Serializable {  private String id;  private String description;  private String category;  private final LocalDateTime date = LocalDateTime.now();  public TodoItem(String id, String description, String category) {   this.id = id;   this.description = description;   this.category = category;  }  public String getId(){   return this.id;  }  public  String getDescription(){   return this.description;  }  public String getCategory(){   return this.category;  }  public void setId(String id) {   this.id = id;  }  public void setDescription(String description) {   this.description = description;  }  public void setCategory(String category) {   this.category = category;  }  @Override  public String toString() {   return  "VALUES ( " + "'" + this.id +"'" + ", " + "'" + this.description +"'" + ", " + "'" + this.category +"'" +", "  + "'" + date +"'" + ")";  } } 

build.gradle清单

apply plugin: 'idea' apply plugin: 'java' apply plugin: 'jetty' apply plugin: 'application' applicationName = 'todocass' applicationDefaultJvmArgs = ["-Dlogback.configurationFile=etc/todosolr/logging.xml"] sourceCompatibility = 1.8 version = '1.0' repositories {  mavenLocal()  mavenCentral() } task runSpark(type: JavaExec, dependsOn: 'classes') {  main = "com.example.SparkApp"  classpath = sourceSets.main.runtimeClasspath } task runSparkSQL(type: JavaExec, dependsOn: 'classes') {  main = "com.example.SparkSQLApp"  classpath = sourceSets.main.runtimeClasspath } dependencies {  //spark and cassandra connector to work with java  compile 'com.datastax.spark:spark-cassandra-connector_2.10:1.1.1-rc4'  compile 'com.datastax.spark:spark-cassandra-connector-java_2.10:1.1.1'  compile 'org.apache.spark:spark-streaming_2.10:1.5.0'  compile 'org.apache.spark:spark-sql_2.10:1.5.0'  //logback dependencies  compile 'ch.qos.logback:logback-core:1.1.3'  compile 'ch.qos.logback:logback-classic:1.1.3'  compile 'org.slf4j:slf4j-api:1.7.12' } //Install/copy tasks task copyDist(type: Copy) {  dependsOn "installApp"  from "$buildDir/install/todocass"  into 'opt/todocass' } task copyLog(type: Copy) {  from "src/main/resources/logback.xml"  into "etc/todocass/" } task copyLogToImage(type: Copy) {  from "src/main/resources/logback.xml"  into "image-todo-cass/etc" } task copyDistToImage(type: Copy) {  dependsOn "installApp"  from "$buildDir/install/todocass"  into "$projectDir/image-todo-cass/opt/todocass" } 

运行

首先运行Cassandra：

cd ~/cassandra bin/cassandra -f

获取代码：

git clone https://github.com/MammatusTech/Spark-Course.git

然后构建Spark-Course：

cd Spark-Course gradle clean build

首先，运行SparkApp，这是Spark与Cassandra协同工作的例子：

gradle runSpark

运行时将看到下列内容：

Query all results from cassandra's todolisttable: [Row[George, Shopping, Mon Jun 15 13:36:07 PDT 2015, Buy a new computer], Row[Janet, Shopping, Mon Jun 15 13:36:07 PDT 2015, buy groceries], Row[John, Sport, Mon Jun 15 13:36:07 PDT 2015, Go to the gym], Row[Paul, Coking, Mon Jun 15 13:36:07 PDT 2015, Prepare lunch], Row[Ron, Education, Mon Jun 15 13:36:07 PDT 2015, Finish the homework], Row[Andy, Fun, Mon Jun 15 13:36:07 PDT 2015, go to the beach], Row[Sam, Shopping, Mon Jun 15 13:36:07 PDT 2015, buy a car]] Saving RDD into a temp table in casssandra then query all results from cassandra: [Row[George, Shopping, Buy a new computer], Row[Janet, Shopping, buy groceries], Row[John, Sport, Go to the gym], Row[Paul, Coking, Prepare lunch], Row[Ron, Education, Finish the homework], Row[Andy, Fun, go to the beach], Row[Sam, Shopping, buy a car]] Reading Data from todolisttable in Cassandra with a RDD: CassandraRow{id: Paul, category: Coking, date: 2015-06-15 13:36:07-0700, description: Prepare lunch} CassandraRow{id: Sam, category: Shopping, date: 2015-06-15 13:36:07-0700, description: buy a car} CassandraRow{id: Ron, category: Education, date: 2015-06-15 13:36:07-0700, description: Finish the homework} CassandraRow{id: Janet, category: Shopping, date: 2015-06-15 13:36:07-0700, description: buy groceries} CassandraRow{id: John, category: Sport, date: 2015-06-15 13:36:07-0700, description: Go to the gym} CassandraRow{id: George, category: Shopping, date: 2015-06-15 13:36:07-0700, description: Buy a new computer} CassandraRow{id: Andy, category: Fun, date: 2015-06-15 13:36:07-0700, description: go to the beach}

然后运行SparkSQL APP，这是Spark SQL与Cassandra协同工作的例子：

gradle runSparkSQL:

运行时将看到下列内容：

Total number of TodoItems = [7] Show the DataFrame result: +---------+-------------------+------+ | category|        description|    id| +---------+-------------------+------+ | Shopping| Buy a new computer|George| |    Sport|      Go to the gym|  John| |Education|Finish the homework|   Ron| | Shopping|          buy a car|   Sam| | Shopping|      buy groceries| Janet| |      Fun|    go to the beach|  Andy| |  Cooking|      Prepare lunch|  Paul| +---------+-------------------+------+ Select the id column and show its contents: +------+ |    id| +------+ |George| |  John| |   Ron| |   Sam| | Janet| |  Andy| |  Paul| +------+

结束语

本文展示了Spark在实时数据分析方面的强大功能，介绍了Spark生态系统中的一个重要部分——Spark SQL和DataFrames。DataFrames构建在RDD之上，RDD的数据分片使得DataFrame能够被并行处理。

在本文中，我们也演示了如何将Spark和Spark SQL与Cassandra结合起来使用。

Apache Spark已经被证明是Hadoop的天然继承，并引领着大数据技术趋势。它在Hadoop生态圈中运行良好，是实现大数据分析的一条快速通道。Spark提供一套易于使用的API，具备大规模分布式任务下的数据分析能力。Apache Spark 使得普通开发者也具备大数据的实时数据分析能力。 Spark SQL是Apache Spark提供的API的一个实例，易于使用而功能强大。

参考文献

•   Spark Docs
•   Spark Github
•   About Spark
•   Mammatus Tech
•   Mammatus Apache Spark Course

原文链接： Analytics with Apache Spark Tutorial Part 2: Spark SQL （译者/牛亚真 审校/朱正贵 责编/仲浩）

译者简介： 牛亚真，本科，2010年毕业于西南大学计算机与信息科学学院信息管理与信息系统专业；研究生，2013年毕业于中国科学院大学文献情报中心情报学专业，计算机信息处理与检索方向。