Bekerja dengan konektor sumber data untuk Apache Spark

Beberapa konektor sumber data Athena tersedia sebagai konektor SparkDSV2. Nama DSV2 konektor Spark memiliki -dsv2 akhiran (misalnya,athena-dynamodb-dsv2).

Berikut ini adalah DSV2 konektor yang tersedia saat ini, nama .format() kelas Spark mereka, dan tautan ke dokumentasi Kueri Federasi Amazon Athena yang sesuai:

DSV2konektor	Spark .format () nama kelas	Dokumentasi
athena-cloudwatch-dsv2	com.amazonaws.athena.connectors.dsv2.cloudwatch.CloudwatchTableProvider	CloudWatch
athena-cloudwatch-metrics-dsv2	com.amazonaws.athena.connectors.dsv2.cloudwatch.metrics.CloudwatchMetricsTableProvider	CloudWatch metrik
athena-aws-cmdb-dsv2	com.amazonaws.athena.connectors.dsv2.aws.cmdb.AwsCmdbTableProvider	CMDB
athena-dynamodb-dsv2	com.amazonaws.athena.connectors.dsv2.dynamodb.DDBTableProvider	DynamoDB

Untuk mengunduh .jar file untuk DSV2 konektor, kunjungi DSV2 GitHub halaman Amazon Athena Query Federation dan lihat Rilis, Rilis <version>, Bagian Aset.

Tentukan toples ke Spark

Untuk menggunakan DSV2 konektor Athena dengan Spark, Anda mengirimkan .jar file untuk konektor ke lingkungan Spark yang Anda gunakan. Bagian berikut menjelaskan kasus-kasus tertentu.

Athena untuk Spark

Untuk informasi tentang menambahkan .jar file kustom dan konfigurasi kustom ke Amazon Athena untuk Apache Spark, lihat. Gunakan properti Spark untuk menentukan konfigurasi kustom

General Spark

Untuk meneruskan .jar file konektor ke Spark, gunakan spark-submit perintah dan tentukan .jar file dalam --jars opsi, seperti pada contoh berikut:

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

Untuk menjalankan spark-submit perintah dengan --jars parameter di AmazonEMR, Anda harus menambahkan langkah ke cluster Amazon EMR Spark Anda. Untuk detail tentang cara menggunakan spark-submit di AmazonEMR, lihat Menambahkan langkah Spark di Panduan EMR Rilis Amazon.

AWS Glue ETLPercikan

Untuk AWS Glue ETL, Anda dapat meneruskan .jar file GitHub .com URL ke --extra-jars argumen aws glue start-job-run perintah. AWS Glue Dokumentasi menjelaskan --extra-jars parameter sebagai mengambil jalur Amazon S3, tetapi parameter juga dapat mengambil. HTTPS URL Untuk informasi selengkapnya, lihat Referensi parameter Job di Panduan AWS Glue Pengembang.

Kueri konektor di Spark

Untuk mengirimkan yang setara dengan kueri federasi Athena yang ada di Apache Spark, gunakan fungsinya. spark.sql() Misalnya, Anda memiliki kueri Athena berikut yang ingin Anda gunakan pada Apache Spark.

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

Untuk melakukan kueri yang sama di Spark menggunakan konektor Amazon Athena DSV2 DynamoDB, gunakan kode berikut:

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

Tentukan parameter

DSV2Versi konektor sumber data Athena menggunakan parameter yang sama dengan konektor sumber data Athena yang sesuai. Untuk informasi parameter, lihat dokumentasi untuk konektor sumber data Athena yang sesuai.

Dalam PySpark kode Anda, gunakan sintaks berikut untuk mengonfigurasi parameter Anda.

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

Misalnya, kode berikut menetapkan parameter konektor DynamoDB Amazon Athena ke. disable_projection_and_casing 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())