EMR Serverless のメタストア設定

Hive メタストアは、スキーマ、パーティション名、データ型など、テーブルに関する構造情報を一元的に保存する場所です。EMR Serverless を使用すると、このテーブルメタデータをジョブにアクセスできるメタストアに保持できます。

Hive メタストアとして 2 つのオプションがあります。

メタストアとしての AWS Glue データカタログの使用

AWS Glue データカタログをメタストアとして使用するように Spark ジョブと Hive ジョブを設定できます。永続的なメタストア、またはさまざまなアプリケーション、サービス、あるいは AWS アカウントで共有されるメタストアが必要である場合は、この設定をお勧めします。データカタログの詳細については、「AWS Glue データカタログの作成」を参照してください。AWS Glue の料金については、「AWS Glue の価格」を参照してください。

EMR Serverless ジョブは、アプリケーションと同じ AWS アカウントまたは別の AWS アカウントで AWS Glue データカタログを使用するように設定できます。

AWS Glue データカタログを設定する

データカタログを設定するには、使用する EMR Serverless アプリケーションのタイプを選択します。

Spark

EMR Studio を使用して EMR Serverless Spark アプリケーションでジョブを実行する場合、AWS Glue データカタログはデフォルトのメタストアです。

SDK または AWS CLI を使用する場合、ジョブ実行の sparkSubmit パラメータで設定 spark.hadoop.hive.metastore.client.factory.class を com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory に設定できます。次の例は、AWS CLI を使用してデータカタログを設定する方法を示しています。

aws emr-serverless start-job-run \
    --application-id application-id \
    --execution-role-arn job-role-arn \
    --job-driver '{
        "sparkSubmit": {
            "entryPoint": "s3://amzn-s3-demo-bucket/code/pyspark/extreme_weather.py",
            "sparkSubmitParameters": "--conf spark.hadoop.hive.metastore.client.factory.class=com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory --conf spark.driver.cores=1 --conf spark.driver.memory=3g --conf spark.executor.cores=4 --conf spark.executor.memory=3g"
        }
    }'

または、Spark コードで新しい SparkSession を作成するときにこの設定をすることができます。

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.appName("SparkSQL")
    .config(
        "spark.hadoop.hive.metastore.client.factory.class",
        "com.amazonaws.glue.catalog.metastore.AWSGlueDataCatalogHiveClientFactory",
    )
    .enableHiveSupport()
    .getOrCreate()
)

# we can query tables with SparkSQL
spark.sql("SHOW TABLES").show()

# we can also them with native Spark
print(spark.catalog.listTables())

Hive

EMR Serverless Hive アプリケーションの場合、データカタログはデフォルトのメタストアです。つまり、EMR Serverless Hive アプリケーションでジョブを実行すると、Hive はアプリケーションと同じ AWS アカウントでデータカタログにメタストア情報を記録します。データカタログをメタストアとして使用するのに、仮想プライベートクラウド (VPC) は必要ありません。

Hive メタストアテーブルにアクセスするには、「AWS Glue の IAM アクセス許可の設定」で説明されている必要な AWS Glue ポリシーを追加します。

EMR Serverless および AWS Glue データカタログのクロスアカウントアクセスを設定する

EMR Serverless のクロスアカウントアクセスを設定するには、まず次の AWS アカウントにサインインする必要があります。

AWS Glue データカタログを使用する場合の考慮事項

Hive スクリプトで ADD JAR に補助 JAR を追加できます。その他の考慮事項については、「AWS Glue Data Catalog を使用する場合の考慮事項」を参照してください。

外部の Hive メタストアの使用

EMR Serverless Spark ジョブと Hive ジョブを設定して、Amazon Aurora や Amazon RDS for MySQL などの外部 Hive メタストアに接続できます。このセクションでは、Amazon RDS Hive メタストアの設定方法、VPC の設定方法、外部メタストアを使用するように EMR Serverless ジョブを設定する方法について説明します。

外部の Hive メタストアを作成する

Spark オプションの設定

JDBC の使用

Amazon RDS for MySQL または Amazon Aurora MySQL インスタンスに基づいて Hive メタストアに接続するように EMR Serverless Spark アプリケーションを設定するには、JDBC 接続を使用します。ジョブ実行の spark-submit パラメータで --jars の付いた mariadb-connector-java.jar を渡します。

aws emr-serverless start-job-run \
  --application-id "application-id" \
  --execution-role-arn "job-role-arn" \
  --job-driver '{
        "sparkSubmit": {
            "entryPoint": "s3://amzn-s3-demo-bucket/scripts/spark-jdbc.py",
            "sparkSubmitParameters": "--jars s3://amzn-s3-demo-bucket/mariadb-connector-java.jar 
            --conf spark.hadoop.javax.jdo.option.ConnectionDriverName=org.mariadb.jdbc.Driver 
            --conf spark.hadoop.javax.jdo.option.ConnectionUserName=<connection-user-name> 
            --conf spark.hadoop.javax.jdo.option.ConnectionPassword=<connection-password>
            --conf spark.hadoop.javax.jdo.option.ConnectionURL=<JDBC-Connection-string> 
            --conf spark.driver.cores=2
            --conf spark.executor.memory=10G 
            --conf spark.driver.memory=6G 
            --conf spark.executor.cores=4"
        }
    }' \
    --configuration-overrides '{
        "monitoringConfiguration": {
        "s3MonitoringConfiguration": {
            "logUri": "s3://amzn-s3-demo-bucket/spark/logs/"
        }
    }
}'

次のコード例は、Amazon RDS の Hive メタストアとやり取りする Spark エントリポイントスクリプトです。

from os.path import expanduser, join, abspath
from pyspark.sql import SparkSession
from pyspark.sql import Row
# warehouse_location points to the default location for managed databases and tables
warehouse_location = abspath('spark-warehouse')
spark = SparkSession \
    .builder \
    .config("spark.sql.warehouse.dir", warehouse_location) \
    .enableHiveSupport() \
    .getOrCreate()
spark.sql("SHOW DATABASES").show()
spark.sql("CREATE EXTERNAL TABLE `sampledb`.`sparknyctaxi`(`dispatching_base_num` string, `pickup_datetime` string, `dropoff_datetime` string, `pulocationid` bigint, `dolocationid` bigint, `sr_flag` bigint) STORED AS PARQUET LOCATION 's3://<s3 prefix>/nyctaxi_parquet/'")
spark.sql("SELECT count(*) FROM sampledb.sparknyctaxi").show()
spark.stop()

thrift サービスの使用

Amazon RDS for MySQL または Amazon Aurora MySQL インスタンスに基づいて Hive メタストアに接続するように EMR Serverless Hive アプリケーションを設定できます。これを行うには、既存の Amazon EMR クラスターのプライマリノードで thrift サーバーを実行します。このオプションは、EMR Serverless ジョブ設定を簡素化するために使用する thrift サーバーを備えた Amazon EMR クラスターが既にある場合に最適です。

aws emr-serverless start-job-run \
  --application-id "application-id" \
  --execution-role-arn "job-role-arn" \
  --job-driver '{
        "sparkSubmit": {
            "entryPoint": "s3://amzn-s3-demo-bucket/thriftscript.py",
            "sparkSubmitParameters": "--jars s3://amzn-s3-demo-bucket/mariadb-connector-java.jar 
            --conf spark.driver.cores=2
            --conf spark.executor.memory=10G 
            --conf spark.driver.memory=6G 
            --conf spark.executor.cores=4"
        }
    }' \
    --configuration-overrides '{
        "monitoringConfiguration": {
        "s3MonitoringConfiguration": {
            "logUri": "s3://amzn-s3-demo-bucket/spark/logs/"
        }
    }
}'

次のコード例は、thrift プロトコルを使用して Hive メタストアに接続するエントリポイントスクリプト (thriftscript.py) です。hive.metastore.uris プロパティは、外部 Hive メタストアから読み取るように設定する必要があります。

from os.path import expanduser, join, abspath
from pyspark.sql import SparkSession
from pyspark.sql import Row
# warehouse_location points to the default location for managed databases and tables
warehouse_location = abspath('spark-warehouse')
spark = SparkSession \
    .builder \
    .config("spark.sql.warehouse.dir", warehouse_location) \
    .config("hive.metastore.uris","thrift://thrift-server-host:thift-server-port") \
    .enableHiveSupport() \
    .getOrCreate()
spark.sql("SHOW DATABASES").show()
spark.sql("CREATE EXTERNAL TABLE sampledb.`sparknyctaxi`( `dispatching_base_num` string, `pickup_datetime` string, `dropoff_datetime` string, `pulocationid` bigint, `dolocationid` bigint, `sr_flag` bigint) STORED AS PARQUET LOCATION 's3://<s3 prefix>/nyctaxi_parquet/'")
spark.sql("SELECT * FROM sampledb.sparknyctaxi").show()
spark.stop()

Hive オプションの設定

JDBC の使用

Amazon RDS MySQL または Amazon Aurora インスタンスのいずれかで外部 Hive データベースの場所を指定する場合は、デフォルトのメタストア設定を上書きできます。

hive-site 分類で次の設定を行って、外部 Hive メタストアをアクティブ化します。

{
    "classification": "hive-site",
    "properties": {
        "hive.metastore.client.factory.class": "org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClientFactory",
        "javax.jdo.option.ConnectionDriverName": "org.mariadb.jdbc.Driver",
        "javax.jdo.option.ConnectionURL": "jdbc:mysql://db-host:db-port/db-name",
        "javax.jdo.option.ConnectionUserName": "username",
        "javax.jdo.option.ConnectionPassword": "password"
    }
}

thrift サーバーの使用

Amazon RDS for MySQL または Amazon Aurora MySQL インスタンスに基づいて Hive メタストアに接続するように EMR Serverless Hive アプリケーションを設定できます。これを行うには、既存の Amazon EMR クラスターのメインノードで thrift サーバーを実行します。このオプションは、thrift サーバーを実行する Amazon EMR クラスターが既にあり、EMR Serverless ジョブ設定を使用する場合に最適です。

EMR Serverless がリモート thrift メタストアにアクセスできるように、hive-site 分類で次の設定を行います。外部 Hive メタストアから読み取るように hive.metastore.uris プロパティを設定する必要があります。

{
    "classification": "hive-site",
    "properties": {
        "hive.metastore.client.factory.class": "org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClientFactory",
        "hive.metastore.uris": "thrift://thrift-server-host:thirft-server-port"
    }
} 

外部メタストアを使用する場合の考慮事項