Spark SQL ジョブ用の AWS Glue Data Catalog のサポート

AWS Glue の Data Catalog は、Apache Hive メタストア互換のカタログです。Data Catalog を外部の Apache Hive メタストアとして使用するように、ジョブと開発エンドポイント AWS Glueを設定できます。その後、Data Catalog に格納されたテーブルに対して Apache Spark SQL クエリを直接実行することができます。AWS Glue ダイナミックフレームは、デフォルトで Data Catalogと 統合されています。また、この機能により Spark SQL ジョブは、Data Catalog を外部の Hive メタストアとして使用できるようになります。

この機能には、AWS Glue API エンドポイントへのネットワークアクセスが必要です。ネットワークアクセスを提供する場合、プライベートサブネットに接続がある AWS Glue ジョブでは VPC エンドポイントまたは NAT ゲートウェイを構成する必要があります。VPCエンドポイントの設定については、「データストアへのネットワークアクセスを設定する」を参照してください。NAT ゲートウェイを作成するには、Amazon VPC ユーザーガイドの「NAT ゲートウェイ」を参照してください。

ジョブ引数と開発エンドポイント引数にそれぞれ "--enable-glue-datacatalog": "" 引数を追加することで、AWS Glue ジョブと開発エンドポイントを設定できます。この引数を渡すことで、Spark が外部 Hive メタストアとして Data Catalog にアクセスできるようにする、特定の設定が構成されます。同時に、SparkSession オブジェクトにある AWS Glue ジョブまたは開発エンドポイント内で作成された Hive のサポートを有効化します。

データ カタログへのアクセスを有効にするには、コンソールの［ジョブの追加］または［エンドポイントの追加］ページのカタログオプション グループで［Hive メタストアとして AWS Glue データ カタログを使用］のチェックボックスをオンにします。ジョブまたは開発エンドポイントに使用される IAM ロールには glue:CreateDatabase アクセス許可が必要です。存在しない場合、「default」というデータベースが Data Catalog に作成されます。

Spark SQL ジョブでこの機能の使用方法の例を見てみましょう。次の例では、s3://awsglue-datasets/examples/us-legislators にある米国国会議員のデータセットをクロールしたと想定しています。

AWS Glue Data Catalog で定義されたテーブルからデータをシリアル化/逆シリアル化するには、Spark ジョブのクラスパスにある AWS Glue Data Catalog で定義された形式の Hive SerDe クラスが、Spark SQL に必要です。

特定の一般的なフォーマットの SerDes は AWS Glue によって配布されています。以下はこれらへの Amazon S3 リンクです。

• JSON

• XML

• Grok

JSON SerDe を追加の JAR として開発エンドポイントに追加します。ジョブでは、引数フィールドで --extra-jars 引数を使用して SerDe を追加できます。詳細については、 を参照してください。AWS Glue ジョブのパラメータ

ここに、Spark SQL 用に Data Catalog を有効にして開発エンドポイントを作成するための入力 JSON の例を示します。

{
    "EndpointName": "Name",
    "RoleArn": "role_ARN",
    "PublicKey": "public_key_contents",
    "NumberOfNodes": 2,
    "Arguments": {
      "--enable-glue-datacatalog": ""
    },
    "ExtraJarsS3Path": "s3://crawler-public/json/serde/json-serde.jar"
}

Spark SQL を使用して、米国国会議員のデータセットから作成されたテーブルにクエリを実行します。

>>> spark.sql("use legislators")
DataFrame[]
>>> spark.sql("show tables").show()
+-----------+------------------+-----------+
|   database|         tableName|isTemporary|
+-----------+------------------+-----------+
|legislators|        areas_json|      false|
|legislators|    countries_json|      false|
|legislators|       events_json|      false|
|legislators|  memberships_json|      false|
|legislators|organizations_json|      false|
|legislators|      persons_json|      false|
+-----------+------------------+-----------+
>>> spark.sql("describe memberships_json").show()
+--------------------+---------+-----------------+
|            col_name|data_type|          comment|
+--------------------+---------+-----------------+
|             area_id|   string|from deserializer|
|     on_behalf_of_id|   string|from deserializer|
|     organization_id|   string|from deserializer|
|                role|   string|from deserializer|
|           person_id|   string|from deserializer|
|legislative_perio...|   string|from deserializer|
|          start_date|   string|from deserializer|
|            end_date|   string|from deserializer|
+--------------------+---------+-----------------+

形式の SerDe クラスがジョブのクラスパスで使用できない場合は、以下に示すようなエラーが表示されます。

>>> spark.sql("describe memberships_json").show()

Caused by: MetaException(message:java.lang.ClassNotFoundException Class org.openx.data.jsonserde.JsonSerDe not found)
    at org.apache.hadoop.hive.metastore.MetaStoreUtils.getDeserializer(MetaStoreUtils.java:399)
    at org.apache.hadoop.hive.ql.metadata.Table.getDeserializerFromMetaStore(Table.java:276)
    ... 64 more

memberships テーブルとは異なる organization_id だけを表示するには、次の SQL クエリを実行します。

>>> spark.sql("select distinct organization_id from memberships_json").show()
+--------------------+
|     organization_id|
+--------------------+
|d56acebe-8fdc-47b...|
|8fa6c3d2-71dc-478...|
+--------------------+

動的フレームで同じ手順を実行する必要がある場合は、以下を実行してください。

>>> memberships = glueContext.create_dynamic_frame.from_catalog(database="legislators", table_name="memberships_json")
>>> memberships.toDF().createOrReplaceTempView("memberships")
>>> spark.sql("select distinct organization_id from memberships").show()
+--------------------+
|     organization_id|
+--------------------+
|d56acebe-8fdc-47b...|
|8fa6c3d2-71dc-478...|
+--------------------+

DynamicFrame は ETL オペレーション用に最適化されていますが、Spark SQL が Data Catalog に直接アクセスできるようにすることで、複雑な SQL ステートメントの実行や、既存のアプリケーションの移植が簡素化されます。