開始
- 開始点: SparkSession
- データフレームの生成
- 無タイプのデータセット操作 (別名、DataFrame操作)
- プログラム的にSQLクエリを実行
- グローバル テンポラリ ビュー
- データセットの生成
- RDDを使った内部操作
- Scalar 関数
- 集約関数
開始点: SparkSession
Sparkの全ての機能へのエントリーポイントはSQLSession
クラスです。基本的な SparkSession
を生成するには、単にSparkSession.builder()
を使います:
import org.apache.spark.sql.SparkSession
val spark = SparkSession
.builder()
.appName("Spark SQL basic example")
.config("spark.some.config.option", "some-value")
.getOrCreate()
Sparkの全ての機能へのエントリーポイントはSQLSession
クラスです。基本的な SparkSession
を生成するには、単にSparkSession.builder()
を使います:
import org.apache.spark.sql.SparkSession;
SparkSession spark = SparkSession
.builder()
.appName("Java Spark SQL basic example")
.config("spark.some.config.option", "some-value")
.getOrCreate();
Sparkの全ての機能へのエントリーポイントはSQLSession
クラスです。基本的な SparkSession
を生成するには、単にSparkSession.builder
を使います:
from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.config("spark.some.config.option", "some-value") \
.getOrCreate()
Sparkの全ての機能へのエントリーポイントはSQLSession
クラスです。基本のSparkSession
を初期化するには、単にsparkR.session()
を呼び出します:
sparkR.session(appName = "R Spark SQL basic example", sparkConfig = list(spark.some.config.option = "some-value"))
初めて起動する時には、sparkR.session()
がグローバルなSparkSession
シングルトン インスタンスを初期化し、続く起動で常にこのインスタンスへのリファレンスを返すことに注意してください。この方法は、ユーザは1度だけSparkSession
を初期化する必要があり、read.df
のSparkRの関数はこのグローバルインスタンスに暗黙的にアクセスすることができ、ユーザはSparkSession
インスタンスをあちこちに渡す必要がありません。
Spark 2.0でのSparkSession
はHiveQLを使ってクエリを書き、Hive UDFにアクセスし、Hiveテーブルからデータを読み込むことができる能力を含むHiveの機能のための組み込みのサポートを提供します。 これらの機能を使うために、既存のHiveのセットアップを持つ必要はありません。
データフレームの生成
SparkSession
を使ってアプリケーションは既存のRDD
、Hiveテーブル、あるいはSpark data sourcesからデータフレームを作成することができます。
例として、以下ではJSONファイルの内容に基づいて、データフレームを生成します。
val df = spark.read.json("examples/src/main/resources/people.json")
// Displays the content of the DataFrame to stdout
df.show()
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
SparkSession
を使ってアプリケーションは既存のRDD
、Hiveテーブル、あるいはSpark data sourcesからデータフレームを作成することができます。
例として、以下ではJSONファイルの内容に基づいて、データフレームを生成します。
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
Dataset<Row> df = spark.read().json("examples/src/main/resources/people.json");
// Displays the content of the DataFrame to stdout
df.show();
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
SparkSession
を使ってアプリケーションは既存のRDD
、Hiveテーブル、あるいはSpark data sourcesからデータフレームを作成することができます。
例として、以下ではJSONファイルの内容に基づいて、データフレームを生成します。
# spark is an existing SparkSession
df = spark.read.json("examples/src/main/resources/people.json")
# Displays the content of the DataFrame to stdout
df.show()
# +----+-------+
# | age| name|
# +----+-------+
# |null|Michael|
# | 30| Andy|
# | 19| Justin|
# +----+-------+
SparkSession
を使って、アプリケーションはローカルのRのdata.frame、Hiveテーブル、あるいはSpark データソースからデータフレームを生成することができます。
例として、以下ではJSONファイルの内容に基づいて、データフレームを生成します。
df <- read.json("examples/src/main/resources/people.json")
# Displays the content of the DataFrame
head(df)
## age name
## 1 NA Michael
## 2 30 Andy
## 3 19 Justin
# Another method to print the first few rows and optionally truncate the printing of long values
showDF(df)
## +----+-------+
## | age| name|
## +----+-------+
## |null|Michael|
## | 30| Andy|
## | 19| Justin|
## +----+-------+
無タイプのデータセット操作 (別名、DataFrame操作)
データフレームは Scala、Java、PythonおよびR での構造化データ操作のためのドメイン固有の言語を提供します。
上で述べたように、Spark 2.0では、データフレームはScalaおよびJava APIでのまさに行のデータセットです。これらのオペレーションは、強く型付けされたScala/Javaのデータセットが付属している "型有りの変換"に対して、"型無しの変換"として参照されます。
以下は、データセットを使った構造化データ処理の幾つかの基本的な例を含んでいます:
// This import is needed to use the $-notation
import spark.implicits._
// Print the schema in a tree format
df.printSchema()
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)
// Select only the "name" column
df.select("name").show()
// +-------+
// | name|
// +-------+
// |Michael|
// | Andy|
// | Justin|
// +-------+
// Select everybody, but increment the age by 1
df.select($"name", $"age" + 1).show()
// +-------+---------+
// | name|(age + 1)|
// +-------+---------+
// |Michael| null|
// | Andy| 31|
// | Justin| 20|
// +-------+---------+
// Select people older than 21
df.filter($"age" > 21).show()
// +---+----+
// |age|name|
// +---+----+
// | 30|Andy|
// +---+----+
// Count people by age
df.groupBy("age").count().show()
// +----+-----+
// | age|count|
// +----+-----+
// | 19| 1|
// |null| 1|
// | 30| 1|
// +----+-----+
データセットで実施できる操作の種類の完全なリストは API ドキュメントを参照してください。
単純なカラムの参照および表現に加えて、データセットは文字列操作、日付計算、一般的な数学操作などを含む関数の豊富なライブラリも持っています。完全なリストはDataFrame 関数リファレンスの中で利用可能です。
// col("...") is preferable to df.col("...")
import static org.apache.spark.sql.functions.col;
// Print the schema in a tree format
df.printSchema();
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)
// Select only the "name" column
df.select("name").show();
// +-------+
// | name|
// +-------+
// |Michael|
// | Andy|
// | Justin|
// +-------+
// Select everybody, but increment the age by 1
df.select(col("name"), col("age").plus(1)).show();
// +-------+---------+
// | name|(age + 1)|
// +-------+---------+
// |Michael| null|
// | Andy| 31|
// | Justin| 20|
// +-------+---------+
// Select people older than 21
df.filter(col("age").gt(21)).show();
// +---+----+
// |age|name|
// +---+----+
// | 30|Andy|
// +---+----+
// Count people by age
df.groupBy("age").count().show();
// +----+-----+
// | age|count|
// +----+-----+
// | 19| 1|
// |null| 1|
// | 30| 1|
// +----+-----+
データセットで実施できる操作の種類の完全なリストは API ドキュメントを参照してください。
単純なカラムの参照および表現に加えて、データセットは文字列操作、日付計算、一般的な数学操作などを含む関数の豊富なライブラリも持っています。完全なリストはDataFrame 関数リファレンスの中で利用可能です。
Pythonでは属性(df.age
) あるいは、インデックス(df['age']
)のどちらかを使ってデータフレームのカラムにアクセスすることができます。データを探索するには前者が便利ですが、後者の形式を使うことをとてもお勧めします。これは将来が保証されており、データフレームクラス上の属性もカラム名を使って破壊しないでしょう。
# spark, df are from the previous example
# Print the schema in a tree format
df.printSchema()
# root
# |-- age: long (nullable = true)
# |-- name: string (nullable = true)
# Select only the "name" column
df.select("name").show()
# +-------+
# | name|
# +-------+
# |Michael|
# | Andy|
# | Justin|
# +-------+
# Select everybody, but increment the age by 1
df.select(df['name'], df['age'] + 1).show()
# +-------+---------+
# | name|(age + 1)|
# +-------+---------+
# |Michael| null|
# | Andy| 31|
# | Justin| 20|
# +-------+---------+
# Select people older than 21
df.filter(df['age'] > 21).show()
# +---+----+
# |age|name|
# +---+----+
# | 30|Andy|
# +---+----+
# Count people by age
df.groupBy("age").count().show()
# +----+-----+
# | age|count|
# +----+-----+
# | 19| 1|
# |null| 1|
# | 30| 1|
# +----+-----+
DataFrameで実施できる操作の種類の完全なリストは API ドキュメントを参照してください。
単純なカラムの参照および表現に加えて、データフレームは文字列操作、日付計算、一般的な数学操作などを含む関数の豊富なライブラリも持っています。完全なリストはDataFrame 関数リファレンスの中で利用可能です。
# Create the DataFrame
df <- read.json("examples/src/main/resources/people.json")
# Show the content of the DataFrame
head(df)
## age name
## 1 NA Michael
## 2 30 Andy
## 3 19 Justin
# Print the schema in a tree format
printSchema(df)
## root
## |-- age: long (nullable = true)
## |-- name: string (nullable = true)
# Select only the "name" column
head(select(df, "name"))
## name
## 1 Michael
## 2 Andy
## 3 Justin
# Select everybody, but increment the age by 1
head(select(df, df$name, df$age + 1))
## name (age + 1.0)
## 1 Michael NA
## 2 Andy 31
## 3 Justin 20
# Select people older than 21
head(where(df, df$age > 21))
## age name
## 1 30 Andy
# Count people by age
head(count(groupBy(df, "age")))
## age count
## 1 19 1
## 2 NA 1
## 3 30 1
DataFrameで実施できる操作の種類の完全なリストは API ドキュメントを参照してください。
単純なカラムの参照および表現に加えて、データフレームは文字列操作、日付計算、一般的な数学操作などを含む関数の豊富なライブラリも持っています。完全なリストはDataFrame 関数リファレンスの中で利用可能です。
プログラム的にSQLクエリを実行
SparkSession
のsql
はアプリケーションがSQLクエリをプログラム的に実行することを可能にし、結果を DataFrame
として返します。
// Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people")
val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
SparkSession
のsql
はアプリケーションがSQLクエリをプログラム的に実行することを可能にし、結果を Dataset<Row>
として返します。
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
// Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people");
Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");
sqlDF.show();
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
SparkSession
のsql
はアプリケーションがSQLクエリをプログラム的に実行することを可能にし、結果を DataFrame
として返します。
# Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people")
sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
# +----+-------+
# | age| name|
# +----+-------+
# |null|Michael|
# | 30| Andy|
# | 19| Justin|
# +----+-------+
sql
関数によってアプリケーションはSQLクエリをプログラム的に実行することができ、結果はSparkDataFrame
として返されます。
df <- sql("SELECT * FROM table")
グローバル テンポラリ ビュー
Spark SQLでのテンポラリビューはセッションスコープで、それを生成したセッションが終了した場合に消えるでしょう。Sparkアプリケーションが終了するまで保持し全てのセッションで共有するテンポラリビューを持ちたい場合は、グローバルテンプレートビューを生成することができます。グローバルテンプレート ビューはシステムが保持するデータベースglobal_temp
に結び付けられ、それを参照するために完全名を使う必要があります。例えば、SELECT * FROM global_temp.view1
。
// Register the DataFrame as a global temporary view
df.createGlobalTempView("people")
// Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show()
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
// Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show()
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
// Register the DataFrame as a global temporary view
df.createGlobalTempView("people");
// Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show();
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
// Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show();
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
# Register the DataFrame as a global temporary view
df.createGlobalTempView("people")
# Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age| name|
# +----+-------+
# |null|Michael|
# | 30| Andy|
# | 19| Justin|
# +----+-------+
# Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show()
# +----+-------+
# | age| name|
# +----+-------+
# |null|Michael|
# | 30| Andy|
# | 19| Justin|
# +----+-------+
データセットの生成
データセットはRDDに似ていますが、Javaのシリアライズ化あるいはKryoを使う代わりに、ネットワークを越えて処理あるいは転送するためにオブジェクトをシリアライズ化するために特別なEncoderを使用します。エンコーダーと標準シリアライズ化のどちらもオブジェクトをバイトに変えることに責任を持ちますが、エンコーダーは動的に生成されるコードであり、Sparkがバイトをオブジェクトへデシリアライズ化無しにフィルタリング、ソーティングおよびハッシュ化のような多くのオペレーションを実施することができる形式を使用します。
case class Person(name: String, age: Long)
// Encoders are created for case classes
val caseClassDS = Seq(Person("Andy", 32)).toDS()
caseClassDS.show()
// +----+---+
// |name|age|
// +----+---+
// |Andy| 32|
// +----+---+
// Encoders for most common types are automatically provided by importing spark.implicits._
val primitiveDS = Seq(1, 2, 3).toDS()
primitiveDS.map(_ + 1).collect() // Returns: Array(2, 3, 4)
// DataFrames can be converted to a Dataset by providing a class. Mapping will be done by name
val path = "examples/src/main/resources/people.json"
val peopleDS = spark.read.json(path).as[Person]
peopleDS.show()
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
import java.util.Arrays;
import java.util.Collections;
import java.io.Serializable;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
public static class Person implements Serializable {
private String name;
private long age;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public long getAge() {
return age;
}
public void setAge(long age) {
this.age = age;
}
}
// Create an instance of a Bean class
Person person = new Person();
person.setName("Andy");
person.setAge(32);
// Encoders are created for Java beans
Encoder<Person> personEncoder = Encoders.bean(Person.class);
Dataset<Person> javaBeanDS = spark.createDataset(
Collections.singletonList(person),
personEncoder
);
javaBeanDS.show();
// +---+----+
// |age|name|
// +---+----+
// | 32|Andy|
// +---+----+
// Encoders for most common types are provided in class Encoders
Encoder<Long> longEncoder = Encoders.LONG();
Dataset<Long> primitiveDS = spark.createDataset(Arrays.asList(1L, 2L, 3L), longEncoder);
Dataset<Long> transformedDS = primitiveDS.map(
(MapFunction<Long, Long>) value -> value + 1L,
longEncoder);
transformedDS.collect(); // Returns [2, 3, 4]
// DataFrames can be converted to a Dataset by providing a class. Mapping based on name
String path = "examples/src/main/resources/people.json";
Dataset<Person> peopleDS = spark.read().json(path).as(personEncoder);
peopleDS.show();
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
RDDを使った内部操作
Spark SQL は既存のRDDをデータセットに変換する二つの異なるメソッドを提供します。1つ目のメソッドはオブジェクトの特定のタイプを含むRDDのスキーマを推測するためにリフレクションを使用します。このリフレクションを基礎としたやり方は、Sparkアプリケーションを書いている時にスキーマを既に知っている場合には、結果としてより簡潔でよく動きます。
データセットを生成する2つ目のメソッドは、スキーマを構築し既存のRDDに適用することができるプログラム的なインタフェースです。このメソッドは多少冗長ですが、カラムとそれらのタイプが実行時まで分からない場合にデータセットを構築することができます。
リフレクションを使ったスキーマの推測
Spark SQLのためのScalaインタフェースは、データフレームへのcaseクラスを含むRDDの自動的な変換をサポートします。caseクラスはテーブルのスキーマを定義します。caseクラスへの引数の名前はリフレクションを使って読み込まれ、カラムの名前となります。case クラスは入れ子にすることもでき、Seq
あるいはArray
のような複雑なタイプを含むことができます。このRDDは明示的にデータフレームに変換され、テーブルとして登録されます。テーブルは後に続くSQL分の中で使うことができます。
// For implicit conversions from RDDs to DataFrames
import spark.implicits._
// Create an RDD of Person objects from a text file, convert it to a Dataframe
val peopleDF = spark.sparkContext
.textFile("examples/src/main/resources/people.txt")
.map(_.split(","))
.map(attributes => Person(attributes(0), attributes(1).trim.toInt))
.toDF()
// Register the DataFrame as a temporary view
peopleDF.createOrReplaceTempView("people")
// SQL statements can be run by using the sql methods provided by Spark
val teenagersDF = spark.sql("SELECT name, age FROM people WHERE age BETWEEN 13 AND 19")
// The columns of a row in the result can be accessed by field index
teenagersDF.map(teenager => "Name: " + teenager(0)).show()
// +------------+
// | value|
// +------------+
// |Name: Justin|
// +------------+
// or by field name
teenagersDF.map(teenager => "Name: " + teenager.getAs[String]("name")).show()
// +------------+
// | value|
// +------------+
// |Name: Justin|
// +------------+
// No pre-defined encoders for Dataset[Map[K,V]], define explicitly
implicit val mapEncoder = org.apache.spark.sql.Encoders.kryo[Map[String, Any]]
// Primitive types and case classes can be also defined as
// implicit val stringIntMapEncoder: Encoder[Map[String, Any]] = ExpressionEncoder()
// row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
teenagersDF.map(teenager => teenager.getValuesMap[Any](List("name", "age"))).collect()
// Array(Map("name" -> "Justin", "age" -> 19))
Spark SQLはJavaBeansのRDDからデータフレームへの自動的な変換をサポートします。 リフレクションを使って取得されたBeanInfo
はテーブルのスキーマを定義します。現在のところ、Spark SQLはMap
フィールドを含むJavaBeansをサポートしません。しかし、入れ子になったJavaBeansと、List
あるいは Array
フィールドはサポートされます。Serializableを実装し全てのフィールドについてgetterおよびsetterを持つクラスを生成することで、JavaBeanを生成することができます。
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.Encoder;
import org.apache.spark.sql.Encoders;
// Create an RDD of Person objects from a text file
JavaRDD<Person> peopleRDD = spark.read()
.textFile("examples/src/main/resources/people.txt")
.javaRDD()
.map(line -> {
String[] parts = line.split(",");
Person person = new Person();
person.setName(parts[0]);
person.setAge(Integer.parseInt(parts[1].trim()));
return person;
});
// Apply a schema to an RDD of JavaBeans to get a DataFrame
Dataset<Row> peopleDF = spark.createDataFrame(peopleRDD, Person.class);
// Register the DataFrame as a temporary view
peopleDF.createOrReplaceTempView("people");
// SQL statements can be run by using the sql methods provided by spark
Dataset<Row> teenagersDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19");
// The columns of a row in the result can be accessed by field index
Encoder<String> stringEncoder = Encoders.STRING();
Dataset<String> teenagerNamesByIndexDF = teenagersDF.map(
(MapFunction<Row, String>) row -> "Name: " + row.getString(0),
stringEncoder);
teenagerNamesByIndexDF.show();
// +------------+
// | value|
// +------------+
// |Name: Justin|
// +------------+
// or by field name
Dataset<String> teenagerNamesByFieldDF = teenagersDF.map(
(MapFunction<Row, String>) row -> "Name: " + row.<String>getAs("name"),
stringEncoder);
teenagerNamesByFieldDF.show();
// +------------+
// | value|
// +------------+
// |Name: Justin|
// +------------+
Spark SQL はデータタイプを推測して、RowオブジェクトのRDDをデータフレームに変換することができます。行はkwargsとしてkey/valueペアをRowクラスに渡すことで構築することができます。このリストのキーは表のカラム名を定義し、JSONファイル上で実施される推測に似たような、データセット全体を標本化することでタイプを推測します。
from pyspark.sql import Row
sc = spark.sparkContext
# Load a text file and convert each line to a Row.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))
# Infer the schema, and register the DataFrame as a table.
schemaPeople = spark.createDataFrame(people)
schemaPeople.createOrReplaceTempView("people")
# SQL can be run over DataFrames that have been registered as a table.
teenagers = spark.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")
# The results of SQL queries are Dataframe objects.
# rdd returns the content as an :class:`pyspark.RDD` of :class:`Row`.
teenNames = teenagers.rdd.map(lambda p: "Name: " + p.name).collect()
for name in teenNames:
print(name)
# Name: Justin
プログラム的なスキーマの指定
事前にケースクラスを定義できない場合(例えば、レコードの構造が文字列にエンコードされている。あるいはテキストのデータセットがパースされ、フィールドが異なるユーザには異なって抽出されるかもしれない)、DataFrame
は3つのステップでプログラム的に生成することができます。
- 元のRDDから
行
のRDDを生成する; - ステップ1で生成したRDD内の
行
の構造に一致するStructType
によって表現されるスキーマを生成する。 - スキーマを
SparkSession
によって提供されるcreateDataFrame
メソッドを使って行
のRDDに適用する。
例えば:
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
// Create an RDD
val peopleRDD = spark.sparkContext.textFile("examples/src/main/resources/people.txt")
// The schema is encoded in a string
val schemaString = "name age"
// Generate the schema based on the string of schema
val fields = schemaString.split(" ")
.map(fieldName => StructField(fieldName, StringType, nullable = true))
val schema = StructType(fields)
// Convert records of the RDD (people) to Rows
val rowRDD = peopleRDD
.map(_.split(","))
.map(attributes => Row(attributes(0), attributes(1).trim))
// Apply the schema to the RDD
val peopleDF = spark.createDataFrame(rowRDD, schema)
// Creates a temporary view using the DataFrame
peopleDF.createOrReplaceTempView("people")
// SQL can be run over a temporary view created using DataFrames
val results = spark.sql("SELECT name FROM people")
// The results of SQL queries are DataFrames and support all the normal RDD operations
// The columns of a row in the result can be accessed by field index or by field name
results.map(attributes => "Name: " + attributes(0)).show()
// +-------------+
// | value|
// +-------------+
// |Name: Michael|
// | Name: Andy|
// | Name: Justin|
// +-------------+
事前にJavaBeanクラスを定義できない場合(例えば、レコードの構造が文字列にエンコードされている。あるいはテキストのデータセットがパースされ、フィールドが異なるユーザには異なって抽出されるかもしれない)、Dataset<Row>
は3つのステップでプログラム的に生成することができます。
- 元のRDDから
行
のRDDを生成する; - ステップ1で生成したRDD内の
行
の構造に一致するStructType
によって表現されるスキーマを生成する。 - スキーマを
SparkSession
によって提供されるcreateDataFrame
メソッドを使って行
のRDDに適用する。
例えば:
import java.util.ArrayList;
import java.util.List;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
// Create an RDD
JavaRDD<String> peopleRDD = spark.sparkContext()
.textFile("examples/src/main/resources/people.txt", 1)
.toJavaRDD();
// The schema is encoded in a string
String schemaString = "name age";
// Generate the schema based on the string of schema
List<StructField> fields = new ArrayList<>();
for (String fieldName : schemaString.split(" ")) {
StructField field = DataTypes.createStructField(fieldName, DataTypes.StringType, true);
fields.add(field);
}
StructType schema = DataTypes.createStructType(fields);
// Convert records of the RDD (people) to Rows
JavaRDD<Row> rowRDD = peopleRDD.map((Function<String, Row>) record -> {
String[] attributes = record.split(",");
return RowFactory.create(attributes[0], attributes[1].trim());
});
// Apply the schema to the RDD
Dataset<Row> peopleDataFrame = spark.createDataFrame(rowRDD, schema);
// Creates a temporary view using the DataFrame
peopleDataFrame.createOrReplaceTempView("people");
// SQL can be run over a temporary view created using DataFrames
Dataset<Row> results = spark.sql("SELECT name FROM people");
// The results of SQL queries are DataFrames and support all the normal RDD operations
// The columns of a row in the result can be accessed by field index or by field name
Dataset<String> namesDS = results.map(
(MapFunction<Row, String>) row -> "Name: " + row.getString(0),
Encoders.STRING());
namesDS.show();
// +-------------+
// | value|
// +-------------+
// |Name: Michael|
// | Name: Andy|
// | Name: Justin|
// +-------------+
事前にkwargsの辞書を定義できない場合(例えば、レコードの構造が文字列にエンコードされている。あるいはテキストのデータセットがパースされ、フィールドが異なるユーザには異なって抽出されるかもしれない)、DataFrame
は3つのステップでプログラム的に生成することができます。
- 元のRDDから組あるいはリストのRDDを生成する;
- ステップ1で生成したRDD内の組あるいはリストの構造に一致する
StructType
によって表現されるスキーマを生成する。 - スキーマを
SparkSession
によって提供されるcreateDataFrame
メソッドを使ってRDDに適用します。
例えば:
# Import data types
from pyspark.sql.types import StringType, StructType, StructField
sc = spark.sparkContext
# Load a text file and convert each line to a Row.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
# Each line is converted to a tuple.
people = parts.map(lambda p: (p[0], p[1].strip()))
# The schema is encoded in a string.
schemaString = "name age"
fields = [StructField(field_name, StringType(), True) for field_name in schemaString.split()]
schema = StructType(fields)
# Apply the schema to the RDD.
schemaPeople = spark.createDataFrame(people, schema)
# Creates a temporary view using the DataFrame
schemaPeople.createOrReplaceTempView("people")
# SQL can be run over DataFrames that have been registered as a table.
results = spark.sql("SELECT name FROM people")
results.show()
# +-------+
# | name|
# +-------+
# |Michael|
# | Andy|
# | Justin|
# +-------+
Scalar 関数
Scalar 関数は、行のグループの値を返す集約関数とは対照的に、行ごとに1つの値を返す関数です。Spark SQL は様々な 組み込みの Scalar 関数 をサポートします。ユーザ定義の Scalar 関数もサポートします。
集約関数
集約関数は、行のグループで1つの値を返す関数です。組み込みの集約関数 はcount()
, count_distinct()
, avg()
, max()
, min()
などのような一般的な集約を提供します。ユーザは定義済みの集約関数に制限されず、独自のそれらを作成することができます。ユーザ定義の集約関数の詳細については、ユーザ定義の集約関数のドキュメントを参照してください。