Supporto del catalogo dati di AWS Glue per i processi Spark SQL

AWS Glue Data Catalog è un catalogo compatibile con il database metastore Apache Hive. Puoi configurare gli endpoint di sviluppo e i processi AWS Glue affinché usino catalogo dati come metastore Apache Hive esterno. Quindi, puoi eseguire le query Apache Spark SQL direttamente sulle tabelle archiviate nel catalogo dati. I frame dinamici AWS Glue si integrano con catalogo dati per impostazione predefinita. Tuttavia con questa caratteristica i processi Spark SQL possono iniziare a usare il catalogo dati come metastore Hive esterno.

Questa caratteristica richiede l'accesso di rete all' endpoint API AWS Glue. Per processi AWS Glue con connessioni situate in sottoreti private, è necessario configurare un endpoint VPC o un gateway NAT, per fornire l'accesso alla rete. Per informazioni sulla configurazione degli endpoint VPC, consulta Impostazione dell'accesso di rete agli archivi di dati. Per creare un gateway NAT, consulta Gateway NAT nella Guida per l'utente di Amazon VPC.

È possibile configurare gli endpoint di sviluppo e i processi AWS Glue aggiungendo l'argomento "--enable-glue-datacatalog": "" agli argomenti degli endpoint di sviluppo e del processo rispettivamente. Questo argomento imposta determinate configurazioni in Spark che gli consentono di accedere al catalogo dati come metastore Hive esterno. Inoltre abilita il supporto Hive nell'oggetto SparkSession creato nell'endpoint di sviluppo o nel processo AWS Glue.

Per abilitare l'accesso del catalogo dati, seleziona la casella di controllo Use AWS Glue Data Catalog as the Hive metastore (Usa AWS Glue Data Catalog come metastore Hive) nel gruppo Catalog options (Opzioni catalogo) nella pagina Add job (Aggiungi processo) o Add endpoint (Aggiungi endpoint) della console. Tieni presente che il ruolo IAM utilizzato per il processo o per l'endpoint di sviluppo deve disporre delle autorizzazioni glue:CreateDatabase. Viene creato un database chiamato "default" nel catalogo dati, nel caso non fosse già presente.

Osserviamo un esempio per utilizzare questa caratteristica nei processi Spark SQL. L'esempio seguente presuppone che hai sottoposto a crawling il set di dati dei legislatori degli Stati Uniti disponibile in s3://awsglue-datasets/examples/us-legislators.

Per serializzare/deserializzare i dati provenienti da tabelle definite in AWS Glue Data Catalog, Spark SQL richiede la classe Hive SerDe per il formato definito nel catalogo dati diAWS Glue nel classpath del processo Spark.

Le classi SerDe per determinati formati comuni sono distribuite da AWS Glue. Di seguito sono elencati i relativi collegamenti Amazon S3:

• JSON

• XML

• Grok

Aggiungi la classe JSON SerDe come un ulteriore JAR all'endpoint di sviluppo. Per i processi, puoi aggiungere la classe SerDe utilizzando l'argomento --extra-jars nel campo degli argomenti. Per ulteriori informazioni, consulta Utilizzo dei parametri del lavoro nei lavori AWS Glue.

Ecco un esempio di JSON di input per creare un endpoint di sviluppo con il catalogo dati abilitato per Spark SQL.

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

Ora esegui una query sulle tabelle create dal set di dati dei legislatori degli Stati Uniti utilizzando 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|
+--------------------+---------+-----------------+

Se la classe SerDe per il formato non è disponibile nel classpath del processo, verrà visualizzato un messaggio di errore simile al seguente.

>>> 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

Per visualizzare solo gli organization_id distinti della tabella memberships, esegui la seguente query SQL.

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

Se è necessaria la stessa operazione per i frame dinamici, esegui la query seguente.

>>> 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...|
+--------------------+

I DynamicFrame oltre a essere ottimizzati per le operazioni ETL, consentendo a Spark SQL di accedere al catalogo dati direttamente, forniscono un modo conciso per eseguire istruzioni SQL complesse o trasferire applicazioni esistenti.