狠狠撸

1
? ??? ???
2016.01.18
???
spark SQL

2
??
? Spark SQL ??
? Tungsten execution engine
? Catalyst optimizer
? RDD, DataFrame
? Dataset
? Spark SQL application (in Java)
? Linking
? example
? ?? ??

3
Spark SQL
? ??/??? ???(structured/semi-structured Data) ??? ???
Spark Library
? Tungsten execution engine & Catalyst optimizer ??
? ??? interface ??
? SQL, HiveSQL queries
? Dataframe API
? Scala, Java, Python, and R?? ?? ??
? Spark 1.3
? Dataset API
? Scala, Java ?? ?? ??
? Spark 1.6
? ??? Input source? ??
? RDD & ?? ???
? JSON ??? ?
? Parquet file
? Hive Table
? ODBC/JCBC ???? ??

4
Tungsten execution engine
? ???? bottleneck??
? I/O? network bandwidth? ??
? High bandwidth SSD & striped HDD? ??
? 10Gbps network? ??
? CPU ? memory?? bottleneck ??? ??
? ???? processing workload? ??
? Disk I/O? ????? ?? input data pruning workload
? Shuffle? ?? serialization? hashing? ?? key bottleneck
? CPU? memory? ??? ???…
? ????? ??? ??? ??? ??? ? ?? System Engine? ??!
? Project Tungsten
? Spark 1.4?? DataFrame? ??
? spark 1.6?? Dataset?? ??
??:
1. Project Tungsten – databrick
2. https://issues.apache.org/jira/browse/SPARK-7075

5
Tungsten execution engine – three Goal
? Memory Management and Binary Processing
? JVM object ??? garbage collection? overhead? ?
=>????? data? ??? ?? Java objects ??? binary format?? ??
? ???? ?? ???? ??? ??? ??
=> denser in-memory data format? ???? ??? ???? ??? ?
? ?? memory accounting (size of bytes) ?? ??(??? Heuristics ??)
? ?? ???? ??? domain semantics? ??? ??? data processing? ???? ?
=>binary format? in memory data? ???? data type? ???? operator? ??
(serialization/deserialization ?? data processing)
? Cache-aware Computation
? sorting and hashing for aggregations, joins, and shuffle? ??? ??? ??
=>memory hierarchy? ???? algorithm? data struncture
? Code Generation
? expression evaluation, DataFrame/SQL operators, serializer? ?? ??? ??
=>modern compilers and CPUs? ??? ??? ??? ? ?? code generation

6
Catalyst optimizer
? ??? ??? ????? ?? ??? ?? ??? extensible optimizer
? extensible design
? ??? optimization techniques? feature?? ??? ??
? ?? ???? optimizer? ???? ???? ??
? Catalyst? ?? (In Spark SQL) ( ??? ??? paper ??)
? Tree ??? ???? optimization
? ?? 4??? ??
?? ??: Catalyst Optimizer - databrick
x+(1+2) ? tree ?? Catalyst? phase

7
RDD, DataFrame
DataFrames / SQL
Structured Binary Data (Tungsten)
? High level relational operation ?? ??
? Catalyst optimization ?? ??
? Lower memory pressure
? Memory accounting (avoid OOMs)
? Faster sorting / hashing / serialization
RDDs
Collections of Native JVM Objects
? ?? ?? ?? data type ??? ??
? Compile-time type-safety ??
? ??? ????? ??
? ? ?? ?? ??? ?? ???..
? ?? ?? cost
? ??? ?? boilerplate(?? ??, ??) ?? ??
? ?? ??? ???? API? ??? ? ????
? Catalyst optimizer & Tungsten execution engine? ??? ??? ? ??? ?
? Domain object? type? ??? ? ?? ?? ??? ? ??? ?
?? ↑??? & ??? ↑

8
Dataset
? RDD? DataFrame? ??? ?? ?? interface API
? http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Dataset.html
? ???
? Fast
? Typesafe
? Support for a variety of object models
? Java Compatible
? Interoperates with DataFrames
Dataset
Structured Binary Data (Tungsten)
? High level relational operation ?? ??
? Catalyst optimization ?? ??
? Lower memory pressure
? Memory accounting (avoid OOMs)
? Faster sorting / hashing / serialization
? ?? ?? ?? data type ??? ??
? Compile-time type-safety ??
? ??? ????? ??
?? ??: technicaltidbit.blogspot.kr/

9
Dataset
? Encoder
? Dataset? ??? ??? Structured Binary data? ??
? JVM object? RDD? ??, DataFrame?? ??? ??
? Processing ? ??? ???? serialization? ??
? RDD/DataFrame type? data? Dataset?? ???? ????
Object? ??? ?? ??? Encoder? ??(?? ????? ???? ??)
? ?? ???? ???? java, kyro Serialization? ?? ?? ???
Data Serialization ?? ?? ?? ??: Introducing Spark Datasets- databrick

10
Dataset
? structured/semi-structured Data ?? => Dataset??
?? RDD DataFrame Dataset
?? ??? ?? ?? ??
??? ?? ?? ?? ??? ???
Type-safety ?? ???? ?? ??
?? ??? ??? ??? ???
RDD-Dataset?
WordCount ???
?? ?? ??
RDD-Dataset?
?? ? memory
??? ??
?? ??: Introducing Spark Datasets- databrick

11
Spark SQL application (in Java)
? Linking
? Pom.xml? ?? ?? ??

12
? sample
examples/src/main/resources/people.json
examples/src/main/resources/people.txt

13
? (DataFrame example 1) Jason File
SparkConf sparkConf = new SparkConf().setAppName("dataFrame");
JavaSparkContext ctx = new JavaSparkContext(sparkConf);
SQLContext sqlContext = new org.apache.spark.sql.SQLContext(ctx);
DataFrame df = sqlContext.read().json("examples/src/main/resources/people.json");
df.show();//1
df.printSchema();//2
df.select("name").show(); //3
df.select(df.col("name"), df.col("age").plus(1)).show(); //4
df.filter(df.col("age").gt(21)).show(); //5
df.groupBy("age").count().show();//6
df.registerTempTable("people");
DataFrame results = sqlContext.sql("SELECT name FROM people");
List<String> names = results.javaRDD().map(new Function<Row, String>() {
public String call(Row row) {return "Name: " + row.getString(0); }
}).collect();
for(String tuple : names){ //7
System.out.println(tuple);
}
ctx.stop();
1.
2.
3. 4.
5.
6.
7.
import java.util.List;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

14
? (DataFrame example 2) Text File : Specifying the Schema
JavaRDD<String> people = ctx.textFile("examples/src/main/resources/people.txt");
String schemaString = "name age";
List<StructField> fields = new ArrayList<StructField>();
for (String fieldName: schemaString.split(" ")) {
fields.add(DataTypes.createStructField(fieldName, DataTypes.StringType, true));
}
StructType schema = DataTypes.createStructType(fields);
JavaRDD<Row> rowRDD = people.map(new Function<String, Row>() {
public Row call(String record) throws Exception {
String[] fields = record.split(",");
return RowFactory.create(fields[0], fields[1].trim());
}
});
DataFrame peopleDataFrame = sqlContext.createDataFrame(rowRDD, schema);
peopleDataFrame.show();//1
peopleDataFrame.printSchema();//2
ctx.stop();
1.
2.

15
? (DataFrame example 3) Text File : Inferring the Schema(JavaBean)
import java.io.Serializable;
public class Person implements Serializable {
private String name;
private int age;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
}

16
? (DataFrame example 3) Text File : Inferring the Schema(JavaBean)
JavaRDD<Person> people = ctx.textFile("examples/src/main/resources/people.txt").map(
new Function<String, Person>() {
public Person call(String line) throws Exception {
String[] parts = line.split(",");
Person person = new Person();
person.setName(parts[0]);
person.setAge(Integer.parseInt(parts[1].trim()));
return person;
}});
DataFrame schemaPeople = sqlContext.createDataFrame(people, Person.class);
schemaPeople.registerTempTable("people");
DataFrame teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19");
teenagers.show();//1
teenagers.printSchema();//2
ctx.stop();
1.
2.
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import java.util.ArrayList;

17
? (Dataset example)
SparkConf sparkConf = new SparkConf().setAppName("dataset");
...
DataFrame schemaPeople = sqlContext.createDataFrame(people, Person.class);
Dataset<Person> schools = schemaPeople.as(Encoders.bean(Person.class));
Dataset<String> strings = schools.map(new BuildString(), Encoders.STRING());
//Dataset<String> strings = schools.map(p-> p.getName()+" is "+ p.getAge()+" years old.", Encoders.STRING());
List<String> result = strings.collectAsList();
for(String tuple : result){//1
System.out.println(tuple);
}
ctx.stop();
class BuildString implements MapFunction<Person, String>
{
public String call(Person p) throws Exception {
return p.getName() + " is " + p.getAge() + "
years old.";
}
}
1.
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;

18
? (Dataset example) Encoder
? http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Encoder.html
? http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Encoders.html
1. Primitive type ???
List<String> data = Arrays.asList("abc", "abc", "xyz");
Dataset<String> ds = context.createDataset(data, Encoders.STRING());
2. tuple type(K,V pair) ???
Encoder<Tuple2<Integer, String>> encoder2 = Encoders.tuple(Encoders.INT(), Encoders.STRING());
List<Tuple2<Integer, String>> data2 = Arrays.asList(new scala.Tuple2(1, "a");
Dataset<Tuple2<Integer, String>> ds2 = context.createDataset(data2, encoder2);
3. Java Beans? ??? reference type ???
Encoders.bean(MyClass.class);

19
?? ??
? Dataset
? https://issues.apache.org/jira/browse/SPARK-9999
? http://spark.apache.org/docs/latest/sql-programming-guide.html
? http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Dataset.html
? http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Encoder.html
? https://databricks.com/blog/2015/11/20/announcing-spark-1-6-preview-in-databricks.html
? https://docs.cloud.databricks.com/docs/spark/1.6/index.html#examples/Dataset%20Aggregator.html
? http://technicaltidbit.blogspot.kr/2015/10/spark-16-datasets-best-of-rdds-and.html
? http://www.slideshare.net/databricks/apache-spark-16-presented-by-databricks-cofounder-patrick-wendell
? Tungsten
? https://issues.apache.org/jira/browse/SPARK-7075
? https://databricks.com/blog/2015/04/28/project-tungsten-bringing-spark-closer-to-bare-metal.html
? catalyst
? https://databricks.com/blog/2015/04/13/deep-dive-into-spark-sqls-catalyst-optimizer.html
? Michael Armbrust et al. Spark SQL: Relational Data Processing in Spark, In SIGMOD , 2015

狠狠撸

Spark sql

Recommended

More Related Content

What's hot (20)

Viewers also liked (20)

Similar to Spark sql (20)

Spark sql