Utilisation de connecteurs de source de données pour Apache Spark

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Certains connecteurs de source de données Athena sont disponibles sous forme de connecteurs Spark DSV2 . Les noms des DSV2 connecteurs Spark ont un -dsv2 suffixe (par exemple,athena-dynamodb-dsv2).

Vous trouverez ci-dessous les DSV2 connecteurs actuellement disponibles, le nom de leur .format() classe Spark et les liens vers leur documentation Amazon Athena Federated Query correspondante :

Pour télécharger .jar des fichiers pour les DSV2 connecteurs, rendez-vous sur la DSV2 GitHub page Amazon Athena Query Federation et consultez la section Releases, Release <version>, Assets.

Spécifiez le fichier jar pour Spark

Pour utiliser les DSV2 connecteurs Athena avec Spark, vous devez envoyer le .jar fichier du connecteur à l'environnement Spark que vous utilisez. Les sections suivantes décrivent les cas spécifiques.

Athena pour Spark

Pour plus d'informations sur l'ajout de fichiers .jar personnalisés et d'une configuration personnalisée dans Amazon Athena pour Apache Spark, consultez. Utiliser les propriétés Spark pour définir une configuration personnalisée

General Spark

Pour transmettre le fichier .jar du connecteur à Spark, utilisez la commande spark-submit et spécifiez le fichier .jar dans l'option --jars, comme dans l'exemple suivant :

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

Pour exécuter une commande spark-submit avec le paramètre --jars sur Amazon EMR, vous devez ajouter une étape à votre cluster Amazon EMR Spark. Pour en savoir plus sur l'utilisation de spark-submit sur Amazon EMR, consultez Ajouter une étape Spark dans le Guide de version d'Amazon EMR.

AWS Glue ETL Spark

Pour l' AWS Glue ETL, vous pouvez transmettre l'URL GitHub .com du .jar fichier à l'--extra-jarsargument de la aws glue start-job-run commande. La AWS Glue documentation décrit le --extra-jars paramètre comme empruntant un chemin Amazon S3, mais il peut également prendre une URL HTTPS. Pour plus d'informations, consultez la référence du paramètre de tâche dans le Guide du développeur AWS Glue .

Interrogez le connecteur sur Spark

Pour envoyer l'équivalent de votre requête fédérée Athena existante sur Apache Spark, utilisez la fonction spark.sql(). Supposons par exemple que vous ayez la requête Athena que vous souhaitez utiliser sur Apache Spark.

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

Pour exécuter la même requête sur Spark à l'aide du DSV2 connecteur Amazon Athena DynamoDB, utilisez le code suivant :

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

Specify parameters

Les DSV2 versions des connecteurs de source de données Athena utilisent les mêmes paramètres que les connecteurs de source de données Athena correspondants. Pour obtenir des informations sur les paramètres, reportez-vous à la documentation du connecteur de source de données Athena correspondant.

Dans votre PySpark code, utilisez la syntaxe suivante pour configurer vos paramètres.

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

Par exemple, le code suivant définit le paramètre disable_projection_and_casing du connecteur Amazon Athena DynamoDB sur 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())