AWS Glue で Parquet 形式を使用する

AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが Parquet データ形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能について説明します。

AWS Glue は Parquet 形式の使用をサポートしています。この形式は、パフォーマンス指向の列ベースのデータ形式です。標準局から発行されている形式の概要については、「Apache Parquet ドキュメントの概要」を参照してください。

AWS Glue を使用して、Amazon S3 およびストリーミングソースから Parquet ファイルを読み取ることができ、Amazon S3 に Parquet ファイルを書き込むこともできます。S3 から、Parquet ファイルが含まれる bzip および gzip アーカイブを読み書きすることができます。このページで説明する設定ではなく、S3 接続パラメータ 上で圧縮動作を設定します。

次の表は、Parquet 形式オプションをサポートする一般的な AWS Glue 機能を示しています。

読み込み	書き込み	ストリーミングの読み取り	小さなファイルのグループ化	ジョブのブックマーク
サポート	サポート対象	サポート	サポートされていません	サポート対象*

^* AWS Glue バージョン 1.0+ でサポート

例: S3 から Parquet ファイルまたはフォルダを読み取る

前提条件: 読み取りたい Parquet ファイルまたはフォルダへの S3 パス (s3path) が必要です。

設定: 関数オプションで format="parquet" を指定します。connection_options で、paths キーを使用して s3path を指定します。

リーダーが S3 とやり取りする方法は、connection_options で設定できます。詳細については、AWS Glue: S3 接続パラメータ の「ETL の接続タイプとオプション」を参照してください。

リーダーが Parquet ファイルを解釈する方法は、format_options で設定できます。詳細については、「Parquet 設定リファレンス」を参照してください。

次の AWS Glue ETL スクリプトは、S3 から Parquet ファイルまたはフォルダを読み取るプロセスを示しています。

Python

この例では create_dynamic_frame.from_options メソッドを使用します。

# Example: Read Parquet from S3

from pyspark.context import SparkContext
from awsglue.context import GlueContext

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)
spark = glueContext.spark_session

dynamicFrame = glueContext.create_dynamic_frame.from_options(
    connection_type = "s3", 
    connection_options = {"paths": ["s3://s3path/"]}, 
    format = "parquet"
)

スクリプト (pyspark.sql.DataFrame) では DataFrame を使用することもできます。

dataFrame = spark.read.parquet("s3://s3path/")
                    

Scala

この例では getSourceWithFormat メソッドを使用します。

// Example: Read Parquet from S3

import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.{DynamicFrame, GlueContext}
import org.apache.spark.SparkContext

object GlueApp {
  def main(sysArgs: Array[String]): Unit = {
    val spark: SparkContext = new SparkContext()
    val glueContext: GlueContext = new GlueContext(spark)
    
    val dynamicFrame = glueContext.getSourceWithFormat(
      connectionType="s3",
      format="parquet",
      options=JsonOptions("""{"paths": ["s3://s3path"]}""")
    ).getDynamicFrame()
  }
}

スクリプト (org.apache.spark.sql.DataFrame) では DataFrame を使用することもできます。

spark.read.parquet("s3://s3path/")

例: Parquet ファイルおよびフォルダを S3 に書き込む

前提条件: 初期化された DataFrame (dataFrame) または DynamicFrame (dynamicFrame) が必要です。また、予想される S3 出力パスである s3path も必要になります。

設定: 関数オプションで format="parquet" を指定します。connection_options で、paths キーを使用して s3path を指定します。

ライターが S3 と対話する方法を、connection_options でさらに詳しく変更することができます。詳細については、AWS Glue: S3 接続パラメータ の「ETL の接続タイプとオプション」を参照してください。操作によってファイルの内容が format_options にどのように書き込まれるかを設定できます。詳細については、「Parquet 設定リファレンス」を参照してください。

次の AWS Glue ETL スクリプトは、S3 に Parquet ファイルとフォルダを書き込むプロセスを示しています。

useGlueParquetWriter 設定キーを通して、DynamicFrame のパフォーマンスを最適化するカスタム Parquet ライターを提供します。このライターがワークロードに適しているかどうかを判断するには、「Glue Parquet Writer」を参照してください。

Python

この例では write_dynamic_frame.from_options メソッドを使用します。

# Example: Write Parquet to S3
# Consider whether useGlueParquetWriter is right for your workflow.

from pyspark.context import SparkContext
from awsglue.context import GlueContext

sc = SparkContext.getOrCreate()
glueContext = GlueContext(sc)

glueContext.write_dynamic_frame.from_options(
    frame=dynamicFrame,
    connection_type="s3",
    format="parquet",
    connection_options={
        "path": "s3://s3path",
    },
    format_options={
        # "useGlueParquetWriter": True,
    },
)

スクリプト (pyspark.sql.DataFrame) では DataFrame を使用することもできます。

df.write.parquet("s3://s3path/")

Scala

この例では getSinkWithFormat メソッドを使用します。

// Example: Write Parquet to S3
// Consider whether useGlueParquetWriter is right for your workflow.

import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.{DynamicFrame, GlueContext}
import org.apache.spark.SparkContext

object GlueApp {
  def main(sysArgs: Array[String]): Unit = {
    val spark: SparkContext = new SparkContext()
    val glueContext: GlueContext = new GlueContext(spark)
    
    glueContext.getSinkWithFormat(
        connectionType="s3",
        options=JsonOptions("""{"path": "s3://s3path"}"""),
        format="parquet"
    ).writeDynamicFrame(dynamicFrame)
  }
}

スクリプト (org.apache.spark.sql.DataFrame) では DataFrame を使用することもできます。

df.write.parquet("s3://s3path/")

Parquet 設定リファレンス

AWS Glue ライブラリが format="parquet" を指定している場合には、以下の format_options を使用することができます。

• useGlueParquetWriter - DynamicFrame ワークフローのパフォーマンスを最適化するカスタム Parquet ライターの使用を指定します。使用方法の詳細については、「Glue Parquet Writer」を参照してください。

  • タイプ: ブール値、デフォルト: false

• compression - 使用する圧縮コーデックを指定します。値は org.apache.parquet.hadoop.metadata.CompressionCodecName との完全な互換性があります。

  • タイプ: 列挙型テキスト、デフォルト: "snappy"

  • 値: "uncompressed"、"snappy"、"gzip"、"lzo"

• blockSize - メモリにバッファされる行グループのサイズを、バイト数で指定します。これはパフォーマンスのチューニングに使用します。サイズは正確に、メガバイト数で分割する必要があります。

  • タイプ: 数値、デフォルト: 134217728

  • デフォルト値は 128 MB です。

• pageSize - ページのサイズをバイト数で指定します。これはパフォーマンスのチューニングに使用します。ページは、単一のレコードにアクセスするために完全に読み取らなければならない最小単位です。

  • タイプ: 数値、デフォルト: 1048576

  • デフォルト値は 1 MB です。

注記

加えて、基盤となる SparkSQL コードで受け入れられるすべてのオプションは、connection_options マップパラメータを介し、この形式に渡されます。例えば、mergeSchema などの Spark 設定を AWS Glue Spark リーダーに行うことで、すべてのファイルのスキーマをマージすることができます。

書き込みパフォーマンスを AWS Glue Parquet ライターで最適化する

注記

AWS Glue Parquet ライターには、これまで glueparquet 形式タイプでアクセスしていました。このアクセス方法は非推奨となりました。代わりに、useGlueParquetWriter を有効にして parquet タイプを使用してください。

AWS Glue Parquet ライターには、Parquet ファイルの書き込みを高速化できるパフォーマンス強化が組み込まれています。これまでのライターでは、書き込む前にスキーマを計算していました。Parquet 形式では、すぐに取得できる形でスキーマを保存しないため、時間がかかることがありました。AWS Glue Parquet ライターを使用すれば、事前に計算済みのスキーマは必要なくなります。データが到着次第、ライターがスキーマを動的に計算して変更します。

useGlueParquetWriter を指定する場合、以下の制限に注意してください。

• ライターは、列の追加や削除などのスキーマの進化のみをサポートし、ResolveChoice のように列タイプの変更はサポートしません。

• ライターは、空の DataFrame の書き込み (スキーマのみのファイルを書き込む場合など) をサポートしていません。enableUpdateCatalog=True の設定によって AWS Glue データカタログと統合する場合、空の DataFrame を書き込もうとしてもデータカタログは更新されません。これにより、スキーマなしでデータカタログにテーブルが作成されます。

変換時にこれらの制限が必要とならない場合、AWS Glue Parquet ライターをオンにすることで、パフォーマンスが向上するはずです。