Membuat Stream melalui API

Gunakan langkah-langkah berikut untuk membuat aliran data Kinesis Anda.

Membangun Klien Kinesis Data Streams

Sebelum Anda dapat bekerja dengan aliran data Kinesis, Anda harus membangun objek klien. Kode Java berikut membuat instance pembuat klien dan menggunakannya untuk mengatur Region, kredensyal, dan konfigurasi klien. Kemudian membangun objek klien.

AmazonKinesisClientBuilder clientBuilder = AmazonKinesisClientBuilder.standard();
        
clientBuilder.setRegion(regionName);
clientBuilder.setCredentials(credentialsProvider);
clientBuilder.setClientConfiguration(config);
        
AmazonKinesis client = clientBuilder.build();

Untuk informasi selengkapnya, lihat Wilayah dan Titik Akhir Kinesis Data Streams di. Referensi Umum AWS

Buat Stream

Sekarang setelah Anda membuat klien Kinesis Data Streams, Anda dapat membuat aliran untuk bekerja dengannya, yang dapat Anda capai dengan konsol Kinesis Data Streams, atau secara terprogram. Untuk membuat aliran secara terprogram, buat instance CreateStreamRequest objek dan tentukan nama untuk aliran dan (jika Anda ingin menggunakan mode yang disediakan) jumlah pecahan untuk aliran yang akan digunakan.

• Sesuai permintaan:

  CreateStreamRequest createStreamRequest = new CreateStreamRequest();
  createStreamRequest.setStreamName( myStreamName );

• Disediakan:

  CreateStreamRequest createStreamRequest = new CreateStreamRequest();
  createStreamRequest.setStreamName( myStreamName );
  createStreamRequest.setShardCount( myStreamSize );

Nama aliran mengidentifikasi aliran. Nama tersebut dicakup ke AWS akun yang digunakan oleh aplikasi. Itu juga dicakup oleh Wilayah. Artinya, dua aliran dalam dua AWS akun berbeda dapat memiliki nama yang sama, dan dua aliran di AWS akun yang sama tetapi di dua Wilayah yang berbeda dapat memiliki nama yang sama, tetapi tidak dua aliran pada akun yang sama dan di Wilayah yang sama.

Throughput aliran adalah fungsi dari jumlah pecahan; lebih banyak pecahan diperlukan untuk throughput yang disediakan lebih besar. Lebih banyak pecahan juga meningkatkan biaya yang AWS dikenakan untuk aliran. Untuk informasi selengkapnya tentang menghitung jumlah pecahan yang sesuai untuk aplikasi Anda, lihatMemilih Mode Kapasitas Aliran Data.

Setelah createStreamRequest objek dikonfigurasi, buat aliran dengan memanggil createStream metode pada klien. Setelah meneleponcreateStream, tunggu streaming mencapai ACTIVE status sebelum melakukan operasi apa pun di aliran. Untuk memeriksa status aliran, panggil describeStream metode. Namun, describeStream melempar pengecualian jika aliran tidak ada. Oleh karena itu, lampirkan describeStream panggilan dalam satu try/catch blok.

client.createStream( createStreamRequest );
DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest();
describeStreamRequest.setStreamName( myStreamName );

long startTime = System.currentTimeMillis();
long endTime = startTime + ( 10 * 60 * 1000 );
while ( System.currentTimeMillis() < endTime ) {
  try {
    Thread.sleep(20 * 1000);
  } 
  catch ( Exception e ) {}
  
  try {
    DescribeStreamResult describeStreamResponse = client.describeStream( describeStreamRequest );
    String streamStatus = describeStreamResponse.getStreamDescription().getStreamStatus();
    if ( streamStatus.equals( "ACTIVE" ) ) {
      break;
    }
    //
    // sleep for one second
    //
    try {
      Thread.sleep( 1000 );
    }
    catch ( Exception e ) {}
  }
  catch ( ResourceNotFoundException e ) {}
}
if ( System.currentTimeMillis() >= endTime ) {
  throw new RuntimeException( "Stream " + myStreamName + " never went active" );
}