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

Migrasi konektor Spark Kinesis ke SDK 2.x untuk Amazon EMR 7.0

AWS SDK menyediakan serangkaian API dan pustaka yang kaya untuk berinteraksi dengan layanan komputasi AWS awan, seperti mengelola kredenSIAL, menghubungkan ke layanan S3 dan Kinesis. Konektor Spark Kinesis digunakan untuk mengkonsumsi data dari Kinesis Data Streams, dan data yang diterima diubah dan diproses dalam mesin eksekusi Spark. Saat ini konektor ini dibangun di atas 1.x AWS SDK dan Kinesis-client-library (KCL).

Sebagai bagian dari migrasi AWS SDK 2.x, konektor Spark Kinesis juga diperbarui sesuai untuk dijalankan dengan SDK 2.x. Dalam rilis Amazon EMR 7.0, Spark berisi peningkatan SDK 2.x yang belum tersedia di versi komunitas Apache Spark. Jika Anda menggunakan konektor Spark Kinesis dari rilis yang lebih rendah dari 7.0, Anda harus memigrasikan kode aplikasi untuk dijalankan di SDK 2.x sebelum Anda dapat bermigrasi ke Amazon EMR 7.0.

Panduan migrasi

Bagian ini menjelaskan langkah-langkah untuk memigrasikan aplikasi ke konektor Spark Kinesis yang ditingkatkan. Ini mencakup panduan untuk bermigrasi ke Kinesis Client Library (KCL) 2.x, penyedia AWS kredensia, dan klien AWS layanan di SDK 2.x. AWS Untuk referensi, ini juga termasuk WordCount program sampel yang menggunakan konektor Kinesis.

Migrasi KCL dari 1.x ke 2.x

Migrating AWS penyedia kredenSIAL dari AWS SDK 1.x hingga 2.x

Penyedia kredenSIAL digunakan untuk mendapatkan AWS kredenSIAL untuk interaksi dengan AWS. Ada beberapa perubahan antarmuka dan kelas yang terkait dengan penyedia kredenSIAL di SDK 2.x, yang dapat ditemukan di sini. Konektor Spark Kinesis telah mendefinisikan antarmuka (org.apache.spark.streaming.kinesis.SparkAWSCredentials) dan kelas implementasi yang mengembalikan versi 1.x dari penyedia AWS kredensia. Penyedia kredenSIAL ini diperlukan saat menginisialisasi klien Kinesis. Misalnya, jika Anda menggunakan metode SparkAWSCredentials.provider dalam aplikasi, Anda perlu memperbarui kode untuk menggunakan penyedia AWS kredensia versi 2.x.

Berikut ini adalah contoh penggunaan penyedia kredensia di AWS SDK 1.x:

import org.apache.spark.streaming.kinesis.SparkAWSCredentials
import com.amazonaws.auth.AWSCredentialsProvider
 
val basicSparkCredentials = SparkAWSCredentials.builder
    .basicCredentials("accessKey", "secretKey")
    .build()
                                     
val credentialProvider = basicSparkCredentials.provider
assert(credentialProvider.isInstanceOf[AWSCredentialsProvider], "Type should be AWSCredentialsProvider")

Migrating AWS Pelayanan Pelanggan dari AWS SDK 1.x hingga 2.x

AWS klien layanan memiliki nama paket yang berbeda di 2.x (yaitusoftware.amazon.awssdk). sedangkan SDK 1.x menggunakan. com.amazonaws Untuk informasi lebih lanjut tentang perubahan klien, lihat di sini. Jika Anda menggunakan klien layanan ini dalam kode, Anda perlu memigrasikan klien yang sesuai.

Berikut ini adalah contoh membuat klien di SDK 1.x:

import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient
import com.amazonaws.services.dynamodbv2.document.DynamoDB
 
AmazonDynamoDB ddbClient = AmazonDynamoDBClientBuilder.defaultClient();
AmazonDynamoDBClient ddbClient = new AmazonDynamoDBClient();

Contoh kode untuk aplikasi streaming

import java.net.URI
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider
import software.amazon.awssdk.http.apache.ApacheHttpClient
import software.amazon.awssdk.services.kinesis.KinesisClient
import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest
import software.amazon.awssdk.regions.Region
import software.amazon.kinesis.metrics.{MetricsLevel, MetricsUtil}
 
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.streaming.dstream.DStream.toPairDStreamFunctions
import org.apache.spark.streaming.kinesis.KinesisInitialPositions.Latest
import org.apache.spark.streaming.kinesis.KinesisInputDStream
 
 
object KinesisWordCountASLSDKV2 {
 
  def main(args: Array[String]): Unit = {
    val appName = "demo-app"
    val streamName = "demo-kinesis-test"
    val endpointUrl = "https://kinesis.us-west-2.amazonaws.com"
    val regionName = "us-west-2"
 
    // Determine the number of shards from the stream using the low-level Kinesis Client
    // from the AWS Java SDK.
    val credentialsProvider = DefaultCredentialsProvider.create
    require(credentialsProvider.resolveCredentials() != null,
      "No AWS credentials found. Please specify credentials using one of the methods specified " +
        "in https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials.html")
    val kinesisClient = KinesisClient.builder()
      .credentialsProvider(credentialsProvider)
      .region(Region.US_WEST_2)
      .endpointOverride(URI.create(endpointUrl))
      .httpClientBuilder(ApacheHttpClient.builder())
      .build()
    val describeStreamRequest = DescribeStreamRequest.builder()
      .streamName(streamName)
      .build()
    val numShards = kinesisClient.describeStream(describeStreamRequest)
      .streamDescription
      .shards
      .size
 
 
    // In this example, we are going to create 1 Kinesis Receiver/input DStream for each shard.
    // This is not a necessity; if there are less receivers/DStreams than the number of shards,
    // then the shards will be automatically distributed among the receivers and each receiver
    // will receive data from multiple shards.
    val numStreams = numShards
 
    // Spark Streaming batch interval
    val batchInterval = Milliseconds(2000)
 
    // Kinesis checkpoint interval is the interval at which the DynamoDB is updated with information
    // on sequence number of records that have been received. Same as batchInterval for this
    // example.
    val kinesisCheckpointInterval = batchInterval
 
    // Setup the SparkConfig and StreamingContext
    val sparkConfig = new SparkConf().setAppName("KinesisWordCountASLSDKV2")
    val ssc = new StreamingContext(sparkConfig, batchInterval)
 
    // Create the Kinesis DStreams
    val kinesisStreams = (0 until numStreams).map { i =>
      KinesisInputDStream.builder
        .streamingContext(ssc)
        .streamName(streamName)
        .endpointUrl(endpointUrl)
        .regionName(regionName)
        .initialPosition(new Latest())
        .checkpointAppName(appName)
        .checkpointInterval(kinesisCheckpointInterval)
        .storageLevel(StorageLevel.MEMORY_AND_DISK_2)
        .metricsLevel(MetricsLevel.DETAILED)
        .metricsEnabledDimensions(Set(MetricsUtil.OPERATION_DIMENSION_NAME, MetricsUtil.SHARD_ID_DIMENSION_NAME))
        .build()
    }
 
    // Union all the streams
    val unionStreams = ssc.union(kinesisStreams)
 
    // Convert each line of Array[Byte] to String, and split into words
    val words = unionStreams.flatMap(byteArray => new String(byteArray).split(" "))
 
    // Map each word to a (word, 1) tuple so we can reduce by key to count the words
    val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
 
    // Print the first 10 wordCounts
    wordCounts.print()
 
    // Start the streaming context and await termination
    ssc.start()
    ssc.awaitTermination()
  }
}

Pertimbangan saat menggunakan konektor Spark Kinesis yang ditingkatkan