Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Koneksi DynamoDB

Anda dapat menggunakan AWS Glue for Spark untuk membaca dan menulis ke tabel di DynamoDB di AWS Glue. Anda terhubung ke DynamoDB menggunakan izin IAM yang dilampirkan ke pekerjaan Glue Anda AWS . AWS Gluemendukung penulisan data ke tabel DynamoDB AWS akun lain. Untuk informasi selengkapnya, lihat Akses lintas akun Lintas wilayah ke tabel DynamoDB.

Konektor DynamoDB asli menggunakan DynamicFrame objek Glue untuk bekerja dengan data yang diekstrak dari DynamoDB. AWS Glue 5.0+ memperkenalkan yang baru Konektor DynamoDB dengan dukungan Spark DataFrame yang menyediakan DataFrame dukungan Spark asli.

Selain konektor AWS Glue DynamoDB ETL, Anda dapat membaca dari DynamoDB menggunakan konektor ekspor DynamoDB, yang memanggil ExportTableToPointInTime permintaan DynamoDB dan menyimpannya di lokasi Amazon S3 yang Anda berikan, dalam format DynamoDB JSON. https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DataExport.Output.html AWS Gluekemudian membuat DynamicFrame objek dengan membaca data dari lokasi ekspor Amazon S3.

Penulis DynamoDB tersedia dalam AWS Glue versi 1.0 atau versi yang lebih baru. Konektor ekspor AWS Glue DynamoDB tersedia dalam AWS Glue versi 2.0 atau versi yang lebih baru. Konektor DataFrame-based DynamoDB baru tersedia dalam AWS Glue versi 5.0 atau versi yang lebih baru.

Untuk informasi selengkapnya tentang DynamoDB, lihat dokumentasi Amazon Dynam oDB.

Mengkonfigurasi koneksi DynamoDB

Untuk menyambung ke DynamoDB dari AWS Glue, berikan peran IAM yang terkait dengan izin pekerjaan AWS Glue Anda untuk berinteraksi dengan DynamoDB. Untuk informasi selengkapnya tentang izin yang diperlukan untuk membaca atau menulis dari DynamoDB, lihat Tindakan, sumber daya, dan kunci kondisi untuk DynamoDB dalam dokumentasi IAM.

Dalam situasi berikut, Anda mungkin memerlukan konfigurasi tambahan:

Membaca dari dan menulis ke DynamoDB

Contoh kode berikut menunjukkan cara membaca dari (melalui konektor ETL) dan menulis ke tabel DynamoDB. Contoh-contoh tersebut menunjukkan pembacaan dari satu tabel dan penulisan ke tabel lain.

Python

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
glue_context= GlueContext(SparkContext.getOrCreate())
job = Job(glue_context)
job.init(args["JOB_NAME"], args)

dyf = glue_context.create_dynamic_frame.from_options(
    connection_type="dynamodb",
    connection_options={"dynamodb.input.tableName": test_source,
        "dynamodb.throughput.read.percent": "1.0",
        "dynamodb.splits": "100"
    }
)
print(dyf.getNumPartitions())

glue_context.write_dynamic_frame_from_options(
    frame=dyf,
    connection_type="dynamodb",
    connection_options={"dynamodb.output.tableName": test_sink,
        "dynamodb.throughput.write.percent": "1.0"
    }
)

job.commit()

Scala

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getSourceWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.input.tableName" -> test_source,
        "dynamodb.throughput.read.percent" -> "1.0",
        "dynamodb.splits" -> "100"
      ))
    ).getDynamicFrame()
    
    print(dynamicFrame.getNumPartitions())

    val dynamoDbSink: DynamoDbDataSink =  glueContext.getSinkWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.output.tableName" -> test_sink,
        "dynamodb.throughput.write.percent" -> "1.0"
      ))
    ).asInstanceOf[DynamoDbDataSink]
    
    dynamoDbSink.writeDynamicFrame(dynamicFrame)

    Job.commit()
  }

}

Menggunakan konektor ekspor DynamoDB

Konektor ekspor berkinerja lebih baik daripada konektor ETL ketika ukuran tabel DynamoDB lebih besar dari 80 GB. Selain itu, mengingat permintaan ekspor dilakukan di luar dari proses Spark dalam suatu AWS Glue pekerjaan, Anda dapat mengaktifkan penskalaan otomatis pekerjaan AWS Glue untuk menyimpan penggunaan DPU selama permintaan ekspor. Dengan konektor ekspor, Anda juga tidak perlu mengonfigurasi jumlah split untuk paralelisme eksekutor Spark atau persentase baca throughput DynamoDB.

Contoh kode berikut menunjukkan cara membaca dari (melalui konektor ekspor) dan mencetak jumlah partisi.

Python

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
glue_context= GlueContext(SparkContext.getOrCreate())
job = Job(glue_context)
job.init(args["JOB_NAME"], args)

dyf = glue_context.create_dynamic_frame.from_options(
    connection_type="dynamodb",
    connection_options={
        "dynamodb.export": "ddb",
        "dynamodb.tableArn": test_source,
        "dynamodb.s3.bucket": bucket_name,
        "dynamodb.s3.prefix": bucket_prefix,
        "dynamodb.s3.bucketOwner": account_id_of_bucket,
    }
)
print(dyf.getNumPartitions())

job.commit()

Scala

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getSourceWithFormat(
      connectionType = "dynamodb",
      options = JsonOptions(Map(
        "dynamodb.export" -> "ddb",
        "dynamodb.tableArn" -> test_source,
        "dynamodb.s3.bucket" -> bucket_name,
        "dynamodb.s3.prefix" -> bucket_prefix,
        "dynamodb.s3.bucketOwner" -> account_id_of_bucket,
      ))
    ).getDynamicFrame()
    
    print(dynamicFrame.getNumPartitions())

    Job.commit()
  }

}

Contoh-contoh ini menunjukkan bagaimana melakukan pembacaan dari (melalui konektor ekspor) dan mencetak jumlah partisi dari tabel Katalog Data AWS Glue yang memiliki dynamodb klasifikasi:

Python

import sys
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions

args = getResolvedOptions(sys.argv, ["JOB_NAME"])
glue_context= GlueContext(SparkContext.getOrCreate())
job = Job(glue_context)
job.init(args["JOB_NAME"], args)

dynamicFrame = glue_context.create_dynamic_frame.from_catalog(
        database=catalog_database,
        table_name=catalog_table_name,
        additional_options={
            "dynamodb.export": "ddb", 
            "dynamodb.s3.bucket": s3_bucket,
            "dynamodb.s3.prefix": s3_bucket_prefix
        }
    )
print(dynamicFrame.getNumPartitions())

job.commit()

Scala

import com.amazonaws.services.glue.GlueContext
import com.amazonaws.services.glue.util.GlueArgParser
import com.amazonaws.services.glue.util.Job
import com.amazonaws.services.glue.util.JsonOptions
import com.amazonaws.services.glue.DynamoDbDataSink
import org.apache.spark.SparkContext
import scala.collection.JavaConverters._


object GlueApp {

  def main(sysArgs: Array[String]): Unit = {
    val glueContext = new GlueContext(SparkContext.getOrCreate())
    val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray)
    Job.init(args("JOB_NAME"), glueContext, args.asJava)
    
    val dynamicFrame = glueContext.getCatalogSource(
        database = catalog_database,
        tableName = catalog_table_name,
        additionalOptions = JsonOptions(Map(
            "dynamodb.export" -> "ddb", 
            "dynamodb.s3.bucket" -> s3_bucket,
            "dynamodb.s3.prefix" -> s3_bucket_prefix
        ))
    ).getDynamicFrame()
    print(dynamicFrame.getNumPartitions())
)

Menyederhanakan penggunaan JSON ekspor DynamoDB

Ekspor DynamoDB dengan konektor ekspor AWS Glue DynamoDB menghasilkan file JSON dari struktur bersarang tertentu. Untuk informasi selengkapnya, lihat Objek data. AWS Gluemenyediakan DynamicFrame transformasi, yang dapat mengubah struktur tersebut menjadi bentuk yang lebih mudah digunakan untuk aplikasi hilir.

Transformasi dapat dipanggil dengan salah satu dari dua cara. Anda dapat mengatur opsi koneksi "dynamodb.simplifyDDBJson" dengan nilai "true" saat memanggil metode untuk membaca dari DynamoDB. Anda juga dapat memanggil transformasi sebagai metode yang tersedia secara independen di perpustakaan AWS Glue.

Pertimbangkan skema berikut yang dihasilkan oleh ekspor DynamoDB:

root
|-- Item: struct
|    |-- parentMap: struct
|    |    |-- M: struct
|    |    |    |-- childMap: struct
|    |    |    |    |-- M: struct
|    |    |    |    |    |-- appName: struct
|    |    |    |    |    |    |-- S: string
|    |    |    |    |    |-- packageName: struct
|    |    |    |    |    |    |-- S: string
|    |    |    |    |    |-- updatedAt: struct
|    |    |    |    |    |    |-- N: string
|    |-- strings: struct
|    |    |-- SS: array
|    |    |    |-- element: string
|    |-- numbers: struct
|    |    |-- NS: array
|    |    |    |-- element: string
|    |-- binaries: struct
|    |    |-- BS: array
|    |    |    |-- element: string
|    |-- isDDBJson: struct
|    |    |-- BOOL: boolean
|    |-- nullValue: struct
|    |    |-- NULL: boolean

simplifyDDBJsonTransformasi akan menyederhanakan ini menjadi:

root
|-- parentMap: struct
|    |-- childMap: struct
|    |    |-- appName: string
|    |    |-- packageName: string
|    |    |-- updatedAt: string
|-- strings: array
|    |-- element: string
|-- numbers: array
|    |-- element: string
|-- binaries: array
|    |-- element: string
|-- isDDBJson: boolean
|-- nullValue: null

Mengkonfigurasi paralelisme dalam operasi DynamoDB

Untuk meningkatkan kinerja, Anda dapat menyetel parameter tertentu yang tersedia untuk konektor DynamoDB. Tujuan Anda saat menyetel parameter paralelisme adalah untuk memaksimalkan penggunaan pekerja Glue yang disediakan. AWS Kemudian, jika Anda membutuhkan lebih banyak kinerja, kami sarankan Anda untuk meningkatkan skala pekerjaan Anda dengan meningkatkan jumlah DPU.

Anda dapat mengubah paralelisme dalam operasi baca DynamoDB menggunakan dynamodb.splits parameter saat menggunakan konektor ETL. Saat membaca dengan konektor ekspor, Anda tidak perlu mengkonfigurasi jumlah split untuk paralelisme eksekutor Spark. Anda dapat mengubah paralelisme dalam operasi penulisan DynamoDB dengan. dynamodb.output.numParallelTasks

Membaca dengan konektor DynamoDB ETL

Kami menyarankan Anda untuk menghitung dynamodb.splits berdasarkan jumlah maksimum pekerja yang ditetapkan dalam konfigurasi pekerjaan Anda dan numSlots perhitungan berikut. Jika penskalaan otomatis, jumlah aktual pekerja yang tersedia dapat berubah di bawah batas itu. Untuk informasi selengkapnya tentang menyetel jumlah maksimum pekerja, lihat Jumlah pekerja (NumberOfWorkers) diMengkonfigurasi properti pekerjaan untuk pekerjaan Spark di AWS Glue.

Kami sarankan Anda dynamodb.splits mengatur jumlah slot yang tersedia,numSlots.

Menulis ke DynamoDB

dynamodb.output.numParallelTasksParameter digunakan untuk menentukan WCU per tugas Spark, menggunakan perhitungan berikut:

permittedWcuPerTask = ( TableWCU * dynamodb.throughput.write.percent ) / dynamodb.output.numParallelTasks

Penulis DynamoDB akan berfungsi paling baik jika konfigurasi secara akurat mewakili jumlah tugas Spark yang menulis ke DynamoDB. Dalam beberapa kasus, Anda mungkin perlu mengganti perhitungan default untuk meningkatkan kinerja penulisan. Jika Anda tidak menentukan parameter ini, tugas WCU yang diizinkan per Spark akan dihitung secara otomatis dengan rumus berikut:

Referensi opsi koneksi DynamoDB

Mengkhususkan koneksi ke Amazon DynamoDB.

Pilihan koneksi berbeda untuk koneksi sumber dan koneksi sink.

“connectionType”: “dynamodb” dengan konektor ETL sebagai sumber

Gunakan opsi koneksi berikut "connectionType": "dynamodb" sebagai sumber, saat menggunakan konektor Glue Dynam AWS oDB ETL:

“connectionType”: “dynamodb” dengan AWS Lem Konektor ekspor DynamoDB sebagai sumber

Gunakan opsi koneksi berikut dengan “ConnectionType”: “dynamodb” sebagai sumber, saat menggunakan konektor ekspor AWS Glue DynamoDB, yang hanya tersedia untuk versi 2.0 dan seterusnya: AWS Glue

“ConnectionType”: “dynamodb” dengan konektor ETL sebagai sink

Gunakan opsi koneksi berikut dengan "connectionType": "dynamodb" sebagai sink: