ORC ファイルの読み取りと書き込み

Apache ORC は、大規模な分析ワークロード用に最適化された列形式のファイル形式です。 組み込みのインデックスと統計を使用して、読み取り中に無関係なデータをスキップします。 Azure Databricksでは、スキーマの指定、パーティション分割、書き込みの圧縮など、Apache Spark での読み取りと書き込みの両方に対して ORC がサポートされています。

前提条件

Azure Databricks では、ORC ファイルを使用するために追加の構成は必要ありません。 ただし、ORC ファイルをストリーミングするには、 自動ローダーが必要です。

オプション

.option() と .options() の DataFrameReader メソッドおよび DataFrameWriter メソッドを使用して、ORC データ ソースを構成します。 サポートされているオプションの完全な一覧は、DataFrameReader ORC オプションおよびDataFrameWriter ORC オプションを参照してください。

Usage

次の例では 、Wanderbricks サンプル データセットを使用して、Spark DataFrame API と SQL を使用した ORC ファイルの読み取りと書き込みを示します。

ORC ファイルの読み取りと書き込み

Python

# Write wanderbricks reviews to ORC format
df = spark.read.table("samples.wanderbricks.reviews")
df.write.format("orc").save("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")

# Read an ORC file into a DataFrame
df = spark.read.format("orc").load("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")
display(df)

# Write with overwrite mode
df.write.format("orc").mode("overwrite").save("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")

Scala

// Write wanderbricks reviews to ORC format
val reviews = spark.read.table("samples.wanderbricks.reviews")
reviews.write.format("orc").save("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")

// Read an ORC file into a DataFrame
val df = spark.read.format("orc").load("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")
df.show()

// Write with overwrite mode
df.write.format("orc").mode("overwrite").save("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")

SQL

-- Write wanderbricks reviews to ORC format
CREATE TABLE reviews_orc
USING ORC
AS SELECT * FROM samples.wanderbricks.reviews;

SELECT * FROM reviews_orc;

SQL を使用して ORC ファイルを読み取る

テーブルを作成せずに SQL を使用してクラウド ストレージから直接 ORC ファイルに対してクエリを実行するには、 read_files を使用します。

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_orc',
  format => 'orc'
)

スキーマを指定する

スキーマ推論のオーバーヘッドを回避するために、ORC ファイルを読み取るときにスキーマを指定します。 たとえば、 review_id、 rating、および comment フィールドを使用してスキーマを定義し、 reviews_orc を DataFrame に読み取ります。

Python

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("review_id", StringType(), True),
    StructField("rating", IntegerType(), True),
    StructField("comment", StringType(), True)
])

df = spark.read.format("orc").schema(schema).load("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")
df.printSchema()
df.show()

Scala

import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

val schema = StructType(Array(
  StructField("review_id", StringType, nullable = true),
  StructField("rating", IntegerType, nullable = true),
  StructField("comment", StringType, nullable = true)
))

val df = spark.read.format("orc").schema(schema).load("/Volumes/<catalog>/<schema>/<volume>/reviews_orc")
df.printSchema()
df.show()

SQL

-- Create a table with an explicit schema from ORC files
CREATE TABLE reviews_orc (
  review_id STRING,
  rating INT,
  comment STRING
)
USING ORC
OPTIONS (path "/Volumes/<catalog>/<schema>/<volume>/reviews_orc");

SELECT * FROM reviews_orc;

パーティション分割された ORC ファイルを書き込む

パーティション分割された ORC ファイルを書き込み、大規模なデータセットのクエリ パフォーマンスを最適化します。 たとえば、samples.wanderbricks.bookings を読み取り、check_in 列から派生した year と month でパーティション分割して bookings_orc_partitioned に書き込みます。

Python

from pyspark.sql.functions import year, month

df = spark.read.table("samples.wanderbricks.bookings")
df_with_parts = df.withColumn("year", year("check_in")).withColumn("month", month("check_in"))
df_with_parts.write.format("orc").partitionBy("year", "month").save("/Volumes/<catalog>/<schema>/<volume>/bookings_orc_partitioned")

Scala

import org.apache.spark.sql.functions.{year, month}

val bookings = spark.read.table("samples.wanderbricks.bookings")
val bookingsWithParts = bookings.withColumn("year", year(col("check_in"))).withColumn("month", month(col("check_in")))
bookingsWithParts.write.format("orc").partitionBy("year", "month").save("/Volumes/<catalog>/<schema>/<volume>/bookings_orc_partitioned")

SQL

-- Write partitioned ORC files by year and month
CREATE TABLE bookings_orc_partitioned
USING ORC
PARTITIONED BY (year, month)
AS SELECT *, year(check_in) AS year, month(check_in) AS month
FROM samples.wanderbricks.bookings;

その他のリソース

  • Azure Databricksの Delta Lake とは何ですか?: ORC を使用して Hive または Hadoop 環境から移行する場合、Delta Lake は Databricks ネイティブ形式として推奨されます。 これは、Parquet ベースのストレージ上に ACID トランザクション、スキーマの強制、時間移動、最適化された読み取りパフォーマンスを追加します。
  • Parquet ファイルの読み取りと書き込み: ワークロードで Databricks 以外で最も広範なエコシステム互換性が必要な場合、Parquet はクエリ エンジンとクラウド ストレージ ツール全体で最も広くサポートされている列形式です。
  • Last updated on