Parquet ファイル
Parquet は多くのほかのデータ処理システムでサポートされているコラム状のフォーマットです。Spark SQL は自動的に元のデータのスキーマを保持するParquetファイルの読み書きの両方のサポートを提供します。Parquetファイルを読み込む場合、互換性の理由から全てのカラムは自動的にnullが可能なように変換されます。
プログラム的なデータのロード
上の例から以下のようにデータを使用します:
// Encoders for most common types are automatically provided by importing spark.implicits._
import spark.implicits._
val peopleDF = spark.read.json("examples/src/main/resources/people.json")
// DataFrames can be saved as Parquet files, maintaining the schema information
peopleDF.write.parquet("people.parquet")
// Read in the parquet file created above
// Parquet files are self-describing so the schema is preserved
// The result of loading a Parquet file is also a DataFrame
val parquetFileDF = spark.read.parquet("people.parquet")
// Parquet files can also be used to create a temporary view and then used in SQL statements
parquetFileDF.createOrReplaceTempView("parquetFile")
val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")
namesDF.map(attributes => "Name: " + attributes(0)).show()
// +------------+
// | value|
// +------------+
// |Name: Justin|
// +------------+
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
Dataset<Row> peopleDF = spark.read().json("examples/src/main/resources/people.json");
// DataFrames can be saved as Parquet files, maintaining the schema information
peopleDF.write().parquet("people.parquet");
// Read in the Parquet file created above.
// Parquet files are self-describing so the schema is preserved
// The result of loading a parquet file is also a DataFrame
Dataset<Row> parquetFileDF = spark.read().parquet("people.parquet");
// Parquet files can also be used to create a temporary view and then used in SQL statements
parquetFileDF.createOrReplaceTempView("parquetFile");
Dataset<Row> namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19");
Dataset<String> namesDS = namesDF.map(
(MapFunction<Row, String>) row -> "Name: " + row.getString(0),
Encoders.STRING());
namesDS.show();
// +------------+
// | value|
// +------------+
// |Name: Justin|
// +------------+
peopleDF = spark.read.json("examples/src/main/resources/people.json")
# DataFrames can be saved as Parquet files, maintaining the schema information.
peopleDF.write.parquet("people.parquet")
# Read in the Parquet file created above.
# Parquet files are self-describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
parquetFile = spark.read.parquet("people.parquet")
# Parquet files can also be used to create a temporary view and then used in SQL statements.
parquetFile.createOrReplaceTempView("parquetFile")
teenagers = spark.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
teenagers.show()
# +------+
# | name|
# +------+
# |Justin|
# +------+
df <- read.df("examples/src/main/resources/people.json", "json")
# SparkDataFrame can be saved as Parquet files, maintaining the schema information.
write.parquet(df, "people.parquet")
# Read in the Parquet file created above. Parquet files are self-describing so the schema is preserved.
# The result of loading a parquet file is also a DataFrame.
parquetFile <- read.parquet("people.parquet")
# Parquet files can also be used to create a temporary view and then used in SQL statements.
createOrReplaceTempView(parquetFile, "parquetFile")
teenagers <- sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19")
head(teenagers)
## name
## 1 Justin
# We can also run custom R-UDFs on Spark DataFrames. Here we prefix all the names with "Name:"
schema <- structType(structField("name", "string"))
teenNames <- dapply(df, function(p) { cbind(paste("Name:", p$name)) }, schema)
for (teenName in collect(teenNames)$name) {
cat(teenName, "\n")
}
## Name: Michael
## Name: Andy
## Name: Justin
パーティションの発見
テーブルのパーティションはHiveのようなシステムで使われる一般的な最適化の方法です。パーティションされたテーブルの中でデータは各パーティションディレクトリのパスでエンコードされたパーティションカラム値を使って、通常異なるディレクトリに保持されます。全ての組み込みファイルソース (Text/CSV/JSON/ORC/Parquetを含む)は自動的にパーティションの情報を発見および推測することができます。例えば、全ての以前に使用したパーティションデータを以下のディレクトリ構造を使って、パーションカラムとしてgender
および country
の2つの追加のカラムを持つパーティションされたテーブルに格納することができます。
path/to/table
を SparkSession.read.parquet
または SparkSession.read.load
のどちらかに渡すことで、Spark SQL は自動的にパスからパーティション情報を抽出することができるでしょう。これで、返されるデータフレームのスキーマは以下のようになります:
パーティションのカラムのデータタイプは自動的に推測されることに注意してください。現在のところ、数学的なデータタイプ、日付、タイムスタンプおよび文字列のタイプがサポートされます。パーティションカラムのデータタイプの自動的な推測をされたくない場合があるかも知れません。そのような場合のために、自動的なタイプの推測はspark.sql.sources.partitionColumnTypeInference.enabled
で設定することができます。デフォルトは true
です。タイプの推測が無効な場合、パーティションカラムとして文字列タイプが使われるでしょう。
Spark 1.6.0から、パーティションの発見はデフォルトで指定されたパスの下のパーティションだけを見つけます。上の例では、もしユーザが path/to/table/gender=male
を SparkSession.read.parquet
または SparkSession.read.load
のどちらかに渡す場合に、gender
はパーティションのカラムとして見なされないでしょう。パーティションの検索を始めるベースパスを指定する必要がある場合は、データソースのオプション内で basePath
を設定することができます。例えば、path/to/table/gender=male
がデータのパスである場合、ユーザはbasePath
を path/to/table/
に設定します。 gender
はパーティションカラムになるでしょう。
スキーマのマージ
Protocol Buffer, Avro および Thrift のように、Parquet もスキーマの評価をサポートします。ユーザは単純なスキーマから開始し、必要に応じて次第にもっとカラムをスキーマに追加することができます。この場合、ユーザは異なるがお互いにスキーマの互換性がある複数のParquetファイルにするかも知れません。Parquetデータソースは現在では自動的にこのケースを検知し、全てのこれらのファイルのスキーマをマージすることができます。
スキーマのマージは比較的高価な操作であり、多くの場合必要ではないため、1.5.0以降からデフォルトではoffにしています。以下のようにして有効にすることができます
- (以下の例のように)Parquet ファイルを読む時にデータソースオプション
mergeSchema
をtrue
に設定、あるいは - グローバルSQLオプション
spark.sql.parquet.mergeSchema
をtrue
に設定。
// This is used to implicitly convert an RDD to a DataFrame.
import spark.implicits._
// Create a simple DataFrame, store into a partition directory
val squaresDF = spark.sparkContext.makeRDD(1 to 5).map(i => (i, i * i)).toDF("value", "square")
squaresDF.write.parquet("data/test_table/key=1")
// Create another DataFrame in a new partition directory,
// adding a new column and dropping an existing column
val cubesDF = spark.sparkContext.makeRDD(6 to 10).map(i => (i, i * i * i)).toDF("value", "cube")
cubesDF.write.parquet("data/test_table/key=2")
// Read the partitioned table
val mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()
// The final schema consists of all 3 columns in the Parquet files together
// with the partitioning column appeared in the partition directory paths
// root
// |-- value: int (nullable = true)
// |-- square: int (nullable = true)
// |-- cube: int (nullable = true)
// |-- key: int (nullable = true)
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
public static class Square implements Serializable {
private int value;
private int square;
// Getters and setters...
}
public static class Cube implements Serializable {
private int value;
private int cube;
// Getters and setters...
}
List<Square> squares = new ArrayList<>();
for (int value = 1; value <= 5; value++) {
Square square = new Square();
square.setValue(value);
square.setSquare(value * value);
squares.add(square);
}
// Create a simple DataFrame, store into a partition directory
Dataset<Row> squaresDF = spark.createDataFrame(squares, Square.class);
squaresDF.write().parquet("data/test_table/key=1");
List<Cube> cubes = new ArrayList<>();
for (int value = 6; value <= 10; value++) {
Cube cube = new Cube();
cube.setValue(value);
cube.setCube(value * value * value);
cubes.add(cube);
}
// Create another DataFrame in a new partition directory,
// adding a new column and dropping an existing column
Dataset<Row> cubesDF = spark.createDataFrame(cubes, Cube.class);
cubesDF.write().parquet("data/test_table/key=2");
// Read the partitioned table
Dataset<Row> mergedDF = spark.read().option("mergeSchema", true).parquet("data/test_table");
mergedDF.printSchema();
// The final schema consists of all 3 columns in the Parquet files together
// with the partitioning column appeared in the partition directory paths
// root
// |-- value: int (nullable = true)
// |-- square: int (nullable = true)
// |-- cube: int (nullable = true)
// |-- key: int (nullable = true)
from pyspark.sql import Row
# spark is from the previous example.
# Create a simple DataFrame, stored into a partition directory
sc = spark.sparkContext
squaresDF = spark.createDataFrame(sc.parallelize(range(1, 6))
.map(lambda i: Row(single=i, double=i ** 2)))
squaresDF.write.parquet("data/test_table/key=1")
# Create another DataFrame in a new partition directory,
# adding a new column and dropping an existing column
cubesDF = spark.createDataFrame(sc.parallelize(range(6, 11))
.map(lambda i: Row(single=i, triple=i ** 3)))
cubesDF.write.parquet("data/test_table/key=2")
# Read the partitioned table
mergedDF = spark.read.option("mergeSchema", "true").parquet("data/test_table")
mergedDF.printSchema()
# The final schema consists of all 3 columns in the Parquet files together
# with the partitioning column appeared in the partition directory paths.
# root
# |-- double: long (nullable = true)
# |-- single: long (nullable = true)
# |-- triple: long (nullable = true)
# |-- key: integer (nullable = true)
df1 <- createDataFrame(data.frame(single=c(12, 29), double=c(19, 23)))
df2 <- createDataFrame(data.frame(double=c(19, 23), triple=c(23, 18)))
# Create a simple DataFrame, stored into a partition directory
write.df(df1, "data/test_table/key=1", "parquet", "overwrite")
# Create another DataFrame in a new partition directory,
# adding a new column and dropping an existing column
write.df(df2, "data/test_table/key=2", "parquet", "overwrite")
# Read the partitioned table
df3 <- read.df("data/test_table", "parquet", mergeSchema = "true")
printSchema(df3)
# The final schema consists of all 3 columns in the Parquet files together
# with the partitioning column appeared in the partition directory paths
## root
## |-- single: double (nullable = true)
## |-- double: double (nullable = true)
## |-- triple: double (nullable = true)
## |-- key: integer (nullable = true)
Hive メタストア Parquet テーブル交換
Hive metastore Parquet テーブルから読み取り、パーティション分割されていない Hive metastore Parquet テーブルに書き込む場合、Spark SQL はパフォーマンスの向上のために Hive SerDe の代わりに独自の Parquet サポートを使おうとします。この挙動は spark.sql.hive.convertMetastoreParquet
設定によって制御され、デフォルトで作動しています。
Hive/Parquet Schema 調整
テーブルスキーマ処理の観点から、HiveとParquetの間には2つの主要な違いがあります。
- Hive は大文字小文字を区別しませんが、Parquetは区別します。
- Hiveは全てのカラムがnull可能ですが、Parquetには重要な意味があります。
この理由により、Hive metastore Parquet テーブルをSpark SQL Parquet テーブルに変換する場合に、Hive metastore スキーマと Parquet スキーマを調停する必要があります。調停ルールは以下の通りです:
-
両方のスキーマで同じ名前を持つフィールドは、null可能かどうかに関係なく同じデータタイプでなければなりません。調停フィールドはParquet側のデータタイプを持たなければなりません。つまりnull可能かどうかが考慮されます。
-
調停スキーマはHive metastoreスキーマで定義されるそれらのフィールドを正確に含まなければなりません。
- Parquetスキーマにのみ現れる全てのフィールドは調停スキーマの中で落とされます。
- Hive metastoreスキーマにのみ現れる全てのフィールドは調停スキーマの中でnull可能なフィールドとして追加されます。
メタデータのリフレッシュ
Spark SQL はパフォーマンスの向上のためにParquet metadetaとしてキャッシュされます。Hive metastore Parquet テーブルの変換が有効な場合、それらの変換されたテーブルのmetadataもキャッシュされます。もしそれらのテーブルがHiveまたは他の外部のツールで更新された場合、metadataの一貫性のために手動でそれらを更新する必要があります。
Columnar Encryption
Spark 3.2以降、Apache Parquet 1.12+を使うParquetテーブルでcolumnar暗号化がサポートされています。
Parquetは、ファイル部分が“データ暗号化キー” (DEKs)で暗号化され、DEKsが“マスター暗号化キー” (MEKs)で暗号化されるエンベロープ暗号化手法を使います。DEKsは暗号化されたファイル/列ごとにParquetによってランダムに生成されます。MEKsはユーザが選択したキー管理サービス(KMS)で生成、格納、管理されます。Parquet Maven リポジトリには、KMSサーバをデプロイせずにspark-shellだけを使って列の暗号化と複合化を実行できるモックKMS実装のjarがあります (parquet-hadoop-tests.jar
ファイルをダウンロードし、それをSpark jars
フォルダに置きます):
KMS Client
InMemoryKMSクラスは、Parquet暗号化機能の説明と簡単なデモンストレーションのためにのみ提供されています。実際の配備では使わないでください。マスター暗号化キーは、ユーザの組織に導入された本番環境グレードのKMSシステムで保持および管理する必要があります。Parquet暗号化を使ったSparkのロールアウトには、KMSサーバのクライアントクラスの実装が必要です。Parquetは、そのようなくらすを開発するための裏グインインタフェースを提供します。
オープンソースのKMSのこのようなクラスの例は、parquet-mrリポジトリにあります。本番のKMSクライアントは組織のセキュリティ管理者と協力して設計し、アクセス制御管理の経験を持つ開発者が構築する必要があります。そのようなクラスが作成されると、上記の暗号化されたデータフレームの書き込み/読み込みのサンプルで示されるように、parquet.encryption.kms.client.class
を介してアプリケーションに渡され、一般的なSparkユーザに利用されることができます。
注意: デフォルトでは、Parquetは“2重エンベロープ暗号化”モードを実装し、Spark executorとKMSサーバとの相互作用を最小限に抑えます。このモードでは、DEKsは“キー暗号化キー” (KEKs、Parquetでランダムに生成されます)で暗号化されます。KEKsはKMSのMEKで暗号化されます; その結果とKEK自体はSpark executorのメモリにキャッシュされます。通常のエンベロープ暗号化に関心のあるユーザは、parquet.encryption.double.wrapping
パラメータをfalse
に設定することで、暗号化に切り替えることができます。Parquet暗号化パラメータの詳細については、parquet-hadoop 設定ページに行ってください。
データソース オプション
Parquetのデータソースオプションは次のやりかたで設定できます:
- the
.option
/.options
methods ofDataFrameReader
DataFrameWriter
DataStreamReader
DataStreamWriter
- CREATE TABLE USING DATA_SOURCEの
OPTIONS
句
プロパティ名 | デフォルト | 意味 | スコープ |
---|---|---|---|
datetimeRebaseMode |
(spark.sql.parquet.datetimeRebaseModeInRead 設定の値) |
The datetimeRebaseMode option allows to specify the rebasing mode for the values of the DATE , TIMESTAMP_MILLIS , TIMESTAMP_MICROS logical types from the Julian to Proleptic Gregorian calendar.現在サポートされるモードは:
|
read |
int96RebaseMode |
(spark.sql.parquet.int96RebaseModeInRead 設定の値) |
int96RebaseMode オプションを使うと、ジュリアン暦から先発グレゴリオ暦までのINT96 タイムスタンプのリベースモードを指定できます。現在サポートされるモードは:
|
read |
mergeSchema |
(spark.sql.parquet.mergeSchema 設定の値) |
全てのParquetパートファイルから収集したスキーマをマージすべきかどうかを設定します。これはspark.sql.parquet.mergeSchema を上書きします。 |
read |
圧縮 |
snappy |
ファイルに保存する時に使う圧縮コーディック。大文字と小文字を区別しない既知の短縮名の1つです(none, uncompressed, snappy, gzip, lzo, brotli, lz4, zstd)。これはspark.sql.parquet.compression.codec を上書きします。 |
write |
そのほかの汎用オプションは、汎用ファイルソースオプションにあります。
設定
Parquetの設定はSparkSession
の setConf
メソッドを使うか、SQLを使ってSET key=value
を実行することで行うことができます。
プロパティ名 | デフォルト | 意味 | これ以降のバージョンから |
---|---|---|---|
spark.sql.parquet.binaryAsString |
false | 他の幾つかのParquet生成システム、特にImpala, Hive および Spark SQLの古いバージョンは、Parquetスキーマを書き出す時にバイナリデータと文字列の区別をしません。このフラグはこれらのシステムとの互換性を提供するために、Spark SQLにバイナリデータを文字列として扱うように指示します。 | 1.1.1 |
spark.sql.parquet.int96AsTimestamp |
true | 幾つかのParquet生成システム、特にImparaおよびHiveは、タイムスタンプをINT96に格納します。このフラグはこれらのシステムとの互換性を提供するために、Spark SQLにINT96データをタイムスタンプとして解釈するように指示します。 | 1.3.0 |
spark.sql.parquet.compression.codec |
snappy |
Parquetファイルを書き込む時に圧縮符号化を使うように設定します。compression または parquet.compression のどちらかがテーブル固有のオプション/プロパティ内で指定された場合、先行詞は compression , parquet.compression , spark.sql.parquet.compression.codec でしょう。利用可能な値には、none, uncompressed, snappy, gzip, lzo, brotli, lz4, zstd が含まれます。zstd はHadoop 2.9.0の前に ZStandardCodec がインストールされることを必要とし、 brotli は BrotliCodec がインストールされることを必要とすることに注意してください。
|
1.1.1 |
spark.sql.parquet.filterPushdown |
true | trueに設定された場合は、Parquet filter push-down 最適化を有効化します。 | 1.2.0 |
spark.sql.hive.convertMetastoreParquet |
true | falseに設定した場合は、Spark SQLはparquetテーブルのためにビルトインサポートの代わりにHive SerDeを使用するでしょう。 | 1.1.1 |
spark.sql.parquet.mergeSchema |
false |
trueの場合、Parquetデータソースは全てのデータファイルから集められたスキーマをマージします。そうでなければ、スキーマは、サマリファイル、あるいはサマリファイルが利用できない場合はランダムデータファイルから取り出されます。 |
1.5.0 |
spark.sql.parquet.writeLegacyFormat |
false | もしtrueであれば、データはSpark1.4以前の方法で書き込まれるでしょう。例えば、小数値はApache Parquetの固定長のバイト配列形式で書き込まれるでしょう。これはApache HiveやApache Impalaのよゆな他のシステムが使います。falseであれば、Parquetの新しい形式が使われるでしょう。例えば、小数はintに基づいた形式で書き込まれるでしょう。Parquetの出力がこの新しい形式をサポートしないシステムと一緒に使うことを意図している場合は、trueに設定してください。 | 1.6.0 |
spark.sql.parquet.datetimeRebaseModeInRead | EXCEPTION |
The rebasing mode for the values of the DATE , TIMESTAMP_MILLIS , TIMESTAMP_MICROS logical types from the Julian to Proleptic Gregorian calendar:
|
3.0.0 |
spark.sql.parquet.datetimeRebaseModeInWrite | EXCEPTION |
The rebasing mode for the values of the DATE , TIMESTAMP_MILLIS , TIMESTAMP_MICROS logical types from the Proleptic Gregorian to Julian calendar:
|
3.0.0 |
spark.sql.parquet.int96RebaseModeInRead | EXCEPTION |
ジュリアン暦から先発グレゴリオ暦までのINT96 タイムスタンプ型の値のリベースモード:
|
3.1.0 |
spark.sql.parquet.int96RebaseModeInWrite | EXCEPTION |
先発グレゴリオ暦からジュリアン暦までのINT96 タイムスタンプ型の値のリベースモード:
|
3.1.0 |