はじめに
- 開始点: SparkSession
- データフレームの作成
- 型指定のないデータセット操作 (別名データフレーム操作)
- SQL クエリをプログラムで実行する
- グローバル一時ビュー
- データセットの作成
- RDD との相互運用
- スカラー関数
- 集約関数
開始点: SparkSession
Spark のすべての機能のエントリポイントは、SparkSession
クラスです。基本的な 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 のすべての機能のエントリポイントは、SparkSession
クラスです。基本的な 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 のすべての機能のエントリポイントは、SparkSession
クラスです。基本的な 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 のすべての機能のエントリポイントは、SparkSession
クラスです。基本的な SparkSession
を初期化するには、sparkR.session()
を呼び出します。
sparkR.session(appName = "R Spark SQL basic example", sparkConfig = list(spark.some.config.option = "some-value"))
sparkR.session()
は、最初に呼び出されたときにグローバルな SparkSession
シングルトンインスタンスを初期化し、以降の呼び出しでは常にこのインスタンスへの参照を返します。このようにして、ユーザーは SparkSession
を一度だけ初期化するだけで、read.df
などの SparkR 関数は暗黙的にこのグローバルインスタンスにアクセスできるようになり、ユーザーは SparkSession
インスタンスを渡す必要がなくなります。
Spark 2.0 の SparkSession
は、HiveQL を使用したクエリ 작성機能、Hive UDF へのアクセス、Hive テーブルからのデータ読み取り機能など、Hive 機能の組み込みサポートを提供します。これらの機能を使用するために、既存の Hive セットアップは必要ありません。
データフレームの作成
SparkSession
を使用すると、アプリケーションは既存の RDD
、Hive テーブル、または Spark データソース からデータフレームを作成できます。
例として、次のコードは 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
を使用すると、アプリケーションは既存の RDD
、Hive テーブル、または Spark データソース からデータフレームを作成できます。
例として、次のコードは 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 データソース からデータフレームを作成できます。
例として、次のコードは 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
を使用すると、アプリケーションはローカルの R データフレーム、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|
## +----+-------+
型指定のないデータセット操作 (別名データフレーム操作)
データフレームは、Scala、Java、Python、および R で構造化データ操作のためのドメイン固有言語を提供します。
前述のように、Spark 2.0 では、データフレームは Scala および Java API の Row
のデータセットです。これらの操作は、厳密に型指定された Scala/Java データセットに付属する「型指定された変換」とは対照的に、「型指定のない変換」とも呼ばれます。
ここでは、データセットを使用した構造化データ処理の基本的な例をいくつか紹介します。
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|
# +----+-----+
データフレームで実行できる操作の種類の完全なリストについては、API ドキュメント を参照してください。
単純な列参照と式に加えて、データフレームには、文字列操作、日付計算、一般的な数学演算などを含む豊富な関数ライブラリがあります。完全なリストは、データフレーム関数リファレンス にあります。
// 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 ドキュメント を参照してください。
単純な列参照と式に加えて、データセットには、文字列操作、日付計算、一般的な数学演算などを含む豊富な関数ライブラリがあります。完全なリストは、データフレーム関数リファレンス にあります。
// 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 ドキュメント を参照してください。
単純な列参照と式に加えて、データセットには、文字列操作、日付計算、一般的な数学演算などを含む豊富な関数ライブラリがあります。完全なリストは、データフレーム関数リファレンス にあります。
# 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
データフレームで実行できる操作の種類の完全なリストについては、API ドキュメント を参照してください。
単純な列参照と式に加えて、データフレームには、文字列操作、日付計算、一般的な数学演算などを含む豊富な関数ライブラリがあります。完全なリストは、データフレーム関数リファレンス にあります。
SQL クエリをプログラムで実行する
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|
# +----+-------+
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|
// +----+-------+
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|
// +----+-------+
CREATE GLOBAL TEMPORARY VIEW temp_view AS SELECT a + 1, b * 2 FROM tbl
SELECT * FROM global_temp.temp_view
データセットの作成
データセットは RDD に似ていますが、Java シリアル化または Kryo を使用するかわりに、特別な エンコーダ を使用して、処理またはネットワーク経由の送信のためにオブジェクトをシリアル化します。エンコーダと標準シリアル化はどちらもオブジェクトをバイトに変換しますが、エンコーダは動的にコード生成され、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 をデータセットに変換するための 2 つの異なる方法をサポートしています。最初の方法は、リフレクションを使用して、特定の種類のオブジェクトを含む RDD のスキーマを推論します。このリフレクションベースのアプローチにより、コードがより簡潔になり、Spark アプリケーションを記述するときにスキーマが既にわかっている場合に適しています。
データセットを作成する 2 番目の方法は、スキーマを構築してから既存の RDD に適用できるプログラムによるインターフェースを使用することです。この方法はより冗長ですが、列とその型が実行時までわからない場合にデータセットを構築できます。
リフレクションを使用したスキーマの推論
Spark SQL は、Row オブジェクトの RDD をデータフレームに変換し、データ型を推論できます。行は、キーと値のペアのリストを kwargs として 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
Spark SQL の Scala インターフェースは、ケースクラスを含む RDD をデータフレームに自動的に変換することをサポートしています。ケースクラスはテーブルのスキーマを定義します。ケースクラスへの引数の名前はリフレクションを使用して読み取られ、列の名前になります。ケースクラスはネストすることも、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をDataFrameに自動的に変換することをサポートしています。リフレクションを使用して取得される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|
// +------------+
スキーマのプログラムによる指定
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|
# +-------+
ケースクラスを事前に定義できない場合(たとえば、レコードの構造が文字列にエンコードされている場合、またはテキストデータセットが解析され、ユーザーごとにフィールドが異なる方法で投影される場合)、DataFrame
は、3つの手順でプログラム的に作成できます。
- 元のRDDから
Row
のRDDを作成します。 - 手順1で作成したRDDの
Row
の構造に一致するStructType
で表されるスキーマを作成します。 SparkSession
によって提供されるcreateDataFrame
メソッドを使用して、スキーマをRow
の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から
Row
のRDDを作成します。 - 手順1で作成したRDDの
Row
の構造に一致するStructType
で表されるスキーマを作成します。 SparkSession
によって提供されるcreateDataFrame
メソッドを使用して、スキーマをRow
の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|
// +-------------+
スカラー関数
スカラー関数は、行ごとに単一の値を返す関数です。これは、行のグループに対して値を返す集約関数とは対照的です。Spark SQLは、さまざまな組み込みスカラー関数をサポートしています。また、ユーザー定義スカラー関数もサポートしています。
集約関数
集約関数は、行のグループに対して単一の値を返す関数です。組み込み集約関数は、count()
、count_distinct()
、avg()
、max()
、min()
などの一般的な集約を提供します。ユーザーは、定義済みの集約関数に限定されず、独自の集約関数を作成できます。ユーザー定義集約関数の詳細については、ユーザー定義集約関数のドキュメントを参照してください。