Uso de conectores de orígenes de datos de Athena para Apache Spark

Algunos conectores de orígenes de datos de Athena están disponibles como conectores DSV2 de Spark. Los nombres de los conectores DSV2 de Spark tienen un sufijo -dsv2 (por ejemplo, athena-dynamodb-dsv2).

A continuación, se muestran los conectores DSV2 disponibles actualmente, su nombre de clase .format() de Spark y los enlaces a la documentación correspondiente de Amazon Athena Federated Query:

Para descargar archivos .jar para los conectores DSV2, visite la página de GitHub Amazon Athena Query Federation DSV2 y consulte la sección Releases (Versiones), Release <version> (Versión <version>), Assets (Activos).

Especificación del archivo jar para Spark

Para usar los conectores DSV2 de Athena con Spark, envíe el archivo .jar del conector al entorno de Spark que esté utilizando. En las siguientes secciones se describen casos específicos.

Athena para Spark

Para obtener información sobre cómo agregar archivos .jar personalizados y configuraciones personalizadas a Amazon Athena para Apache Spark, consulte Uso de las propiedades de Spark para especificar una configuración personalizada.

Spark general

Para pasar el archivo .jar del conector a Spark, utilice el comando spark-submit y especifique el archivo .jar en la opción --jars, como en el siguiente ejemplo:

spark-submit \ 
  --deploy-mode cluster \ 
  --jars https://github.com/awslabs/aws-athena-query-federation-dsv2/releases/download/some_version/athena-dynamodb-dsv2-some_version.jar

Amazon EMR Spark

Para ejecutar un comando spark-submit con el parámetro --jars en Amazon EMR, debe agregar un paso a su clúster de Amazon EMR Spark. Para obtener más información sobre cómo usar spark-submit en Amazon EMR, consulte Add a Spark step (Agregar un paso de Spark) en la Guía de publicación de Amazon EMR.

ETL de AWS Glue para Spark

En el caso de ETL de AWS Glue, puede pasar la URL de GitHub.com del archivo .jar al argumento --extra-jars del comando aws glue start-job-run. En la documentación de AWS Glue, se describe que el parámetro --extra-jars toma una ruta de Amazon S3, pero el parámetro también puede tomar una URL HTTPS. Para obtener más información, consulte Referencia de parámetros de trabajos en la Guía para desarrolladores de AWS Glue.

Consulta del conector en Spark

Para enviar el equivalente a su consulta federada de Athena existente en Apache Spark, use la función spark.sql(). Por ejemplo, supongamos que tiene la siguiente consulta de Athena que desea utilizar en Apache Spark.

SELECT somecola, somecolb, somecolc 
FROM ddb_datasource.some_schema_or_glue_database.some_ddb_or_glue_table 
WHERE somecola > 1

Para realizar la misma consulta en Spark mediante el conector DynamoDB DSV2 de Amazon Athena, utilice el siguiente código:

dynamoDf = (spark.read 
    .option("athena.connectors.schema", "some_schema_or_glue_database") 
    .option("athena.connectors.table", "some_ddb_or_glue_table") 
    .format("com.amazonaws.athena.connectors.dsv2.dynamodb.DDBTableProvider") 
    .load()) 
 
dynamoDf.createOrReplaceTempView("ddb_spark_table") 
 
spark.sql(''' 
SELECT somecola, somecolb, somecolc 
FROM ddb_spark_table 
WHERE somecola > 1 
''')

Especificación de parámetros

Las versiones DSV2 de los conectores de orígenes de datos de Athena utilizan los mismos parámetros que los conectores de orígenes de datos de Athena correspondientes. Para obtener información sobre los parámetros, consulte la documentación del conector de origen de datos de Athena correspondiente.

En su código de PySpark, use la siguiente sintaxis para configurar sus parámetros.

spark.read.option("athena.connectors.conf.parameter", "value")

Por ejemplo, el siguiente código establece el parámetro disable_projection_and_casing del conector DynamoDB de Amazon Athena en always.

dynamoDf = (spark.read 
    .option("athena.connectors.schema", "some_schema_or_glue_database") 
    .option("athena.connectors.table", "some_ddb_or_glue_table") 
    .option("athena.connectors.conf.disable_projection_and_casing", "always") 
    .format("com.amazonaws.athena.connectors.dsv2.dynamodb.DDBTableProvider") 
    .load())