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|
// +------------+
例の完全なコードは Spark のリポジトリの "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" で見つかります。
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|
// +------------+
例の完全なコードは Spark のリポジトリの "examples/src/main/java/org/apache/spark/examples/sql/JavaSQLDataSourceExample.java" で見つかります。
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|
# +------+
例の完全なコードは Spark のリポジトリの "examples/src/main/python/sql/datasource.py" で見つかります。
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
例の完全なコードは Spark のリポジトリの "examples/src/main/r/RSparkSQLExample.R" で見つかります。
CREATE TEMPORARY VIEW parquetTable
USING org.apache.spark.sql.parquet
OPTIONS (
  path "examples/src/main/resources/people.parquet"
)

SELECT * FROM parquetTable

パーティションの発見

テーブルのパーティションはHiveのようなシステムで使われる一般的な最適化の方法です。パーティションされたテーブルの中でデータは各パーティションディレクトリのパスでエンコードされたパーティションカラム値を使って、通常異なるディレクトリに保持されます。全ての組み込みファイルソース (Text/CSV/JSON/ORC/Parquetを含む)は自動的にパーティションの情報を発見および推測することができます。例えば、全ての以前に使用したパーティションデータを以下のディレクトリ構造を使って、パーションカラムとしてgender および country の2つの追加のカラムを持つパーティションされたテーブルに格納することができます。

path
└── to
    └── table
        ├── gender=male
        │   ├── ...
        │   │
        │   ├── country=US
        │   │   └── data.parquet
        │   ├── country=CN
        │   │   └── data.parquet
        │   └── ...
        └── gender=female
            ├── ...
            │
            ├── country=US
            │   └── data.parquet
            ├── country=CN
            │   └── data.parquet
            └── ...

path/to/tableSparkSession.read.parquet または SparkSession.read.loadのどちらかに渡すことで、Spark SQL は自動的にパスからパーティション情報を抽出することができるでしょう。これで、返されるデータフレームのスキーマは以下のようになります:

root
|-- name: string (nullable = true)
|-- age: long (nullable = true)
|-- gender: string (nullable = true)
|-- country: string (nullable = true)

パーティションのカラムのデータタイプは自動的に推測されることに注意してください。現在のところ、数学的なデータタイプ、日付、タイムスタンプおよび文字列のタイプがサポートされます。パーティションカラムのデータタイプの自動的な推測をされたくない場合があるかも知れません。そのような場合のために、自動的なタイプの推測はspark.sql.sources.partitionColumnTypeInference.enabledで設定することができます。デフォルトは trueです。タイプの推測が無効な場合、パーティションカラムとして文字列タイプが使われるでしょう。

Spark 1.6.0から、パーティションの発見はデフォルトで指定されたパスの下のパーティションだけを見つけます。上の例では、もしユーザが path/to/table/gender=maleSparkSession.read.parquet または SparkSession.read.loadのどちらかに渡す場合に、gender はパーティションのカラムとして見なされないでしょう。パーティションの検索を始めるベースパスを指定する必要がある場合は、データソースのオプション内で basePathを設定することができます。例えば、path/to/table/gender=maleがデータのパスである場合、ユーザはbasePathpath/to/table/に設定します。 gender はパーティションカラムになるでしょう。

スキーマのマージ

Protocol Buffer, Avro および Thrift のように、Parquet もスキーマの評価をサポートします。ユーザは単純なスキーマから開始し、必要に応じて次第にもっとカラムをスキーマに追加することができます。この場合、ユーザは異なるがお互いにスキーマの互換性がある複数のParquetファイルにするかも知れません。Parquetデータソースは現在では自動的にこのケースを検知し、全てのこれらのファイルのスキーマをマージすることができます。

スキーマのマージは比較的高価な操作であり、多くの場合必要ではないため、1.5.0以降からデフォルトではoffにしています。以下のようにして有効にすることができます

  1. (以下の例のように)Parquet ファイルを読む時にデータソースオプションmergeSchematrue に設定、あるいは
  2. グローバルSQLオプションspark.sql.parquet.mergeSchematrueに設定。
// 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)
例の完全なコードは Spark のリポジトリの "examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala" で見つかります。
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)
例の完全なコードは Spark のリポジトリの "examples/src/main/java/org/apache/spark/examples/sql/JavaSQLDataSourceExample.java" で見つかります。
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)
例の完全なコードは Spark のリポジトリの "examples/src/main/python/sql/datasource.py" で見つかります。
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)
例の完全なコードは Spark のリポジトリの "examples/src/main/r/RSparkSQLExample.R" で見つかります。

Hive メタストア Parquet テーブル交換

Hive metastore Parquet テーブルから読み取り、パーティション分割されていない Hive metastore Parquet テーブルに書き込む場合、Spark SQL はパフォーマンスの向上のために Hive SerDe の代わりに独自の Parquet サポートを使おうとします。この挙動は spark.sql.hive.convertMetastoreParquet設定によって制御され、デフォルトで作動しています。

Hive/Parquet Schema 調整

テーブルスキーマ処理の観点から、HiveとParquetの間には2つの主要な違いがあります。

  1. Hive は大文字小文字を区別しませんが、Parquetは区別します。
  2. Hiveは全てのカラムがnull可能ですが、Parquetには重要な意味があります。

この理由により、Hive metastore Parquet テーブルをSpark SQL Parquet テーブルに変換する場合に、Hive metastore スキーマと Parquet スキーマを調停する必要があります。調停ルールは以下の通りです:

  1. 両方のスキーマで同じ名前を持つフィールドは、null可能かどうかに関係なく同じデータタイプでなければなりません。調停フィールドはParquet側のデータタイプを持たなければなりません。つまりnull可能かどうかが考慮されます。

  2. 調停スキーマはHive metastoreスキーマで定義されるそれらのフィールドを正確に含まなければなりません。

    • Parquetスキーマにのみ現れる全てのフィールドは調停スキーマの中で落とされます。
    • Hive metastoreスキーマにのみ現れる全てのフィールドは調停スキーマの中でnull可能なフィールドとして追加されます。

メタデータのリフレッシュ

Spark SQL はパフォーマンスの向上のためにParquet metadetaとしてキャッシュされます。Hive metastore Parquet テーブルの変換が有効な場合、それらの変換されたテーブルのmetadataもキャッシュされます。もしそれらのテーブルがHiveまたは他の外部のツールで更新された場合、metadataの一貫性のために手動でそれらを更新する必要があります。

// spark is an existing SparkSession
spark.catalog.refreshTable("my_table")
// spark is an existing SparkSession
spark.catalog().refreshTable("my_table");
# spark is an existing SparkSession
spark.catalog.refreshTable("my_table")
refreshTable("my_table")
REFRESH TABLE my_table;

設定

Parquetの設定はSparkSessionsetConf メソッドを使うか、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 がインストールされることを必要とし、 brotliBrotliCodec がインストールされることを必要とすることに注意してください。 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
TOP
inserted by FC2 system