MSK

Objek yang menggambarkan tipe sumber peristiwa MSK. Untuk informasi selengkapnya, lihat Menggunakan AWS Lambda MSK Amazon di Panduan AWS Lambda Pengembang.

AWS Serverless Application Model (AWS SAM) menghasilkan AWS::Lambda::EventSourceMappingsumber daya saat jenis acara ini disetel.

Untuk menggunakan Schema Registry, Anda perlu menentukan izin peran IAM tertentu untuk fungsi Anda. Lihat Penyiapan lengkap dengan peran IAM untuk contoh konfigurasi yang diperlukan.

Sintaksis

Untuk mendeklarasikan entitas ini di AWS SAM template Anda, gunakan sintaks berikut.

YAML


  BatchSize: Integer
  ConsumerGroupId: String
  DestinationConfig: DestinationConfig
  FilterCriteria: FilterCriteria
  KmsKeyArn: String 
  MaximumBatchingWindowInSeconds: Integer
  ProvisionedPollerConfig: ProvisionedPollerConfig
  SchemaRegistryConfig: SchemaRegistryConfig
  SourceAccessConfigurations: SourceAccessConfigurations
  StartingPosition: String
  StartingPositionTimestamp: Double
  Stream: String
  Topics: List

Properti

BatchSize

Jumlah maksimum rekaman di setiap batch yang Lambda tarik dari aliran atau antrian Anda dan dikirim ke fungsi Anda. Lambda meneruskan semua catatan dalam batch ke fungsi dalam satu panggilan, hingga batas payload untuk pemanggilan sinkron (6 MB).

Default: 100

Rentang Valid: Nilai minimum 1. Nilai maksimum 10.000.

Tipe: Integer

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke BatchSize properti AWS::Lambda::EventSourceMapping sumber daya.

ConsumerGroupId

String yang mengonfigurasi bagaimana acara akan dibaca dari topik Kafka.

Tipe: String

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke AmazonManagedKafkaConfiguration properti AWS::Lambda::EventSourceMapping sumber daya.

DestinationConfig

Objek konfigurasi yang menentukan tujuan dari peristiwa setelah Lambda memprosesnya.

Gunakan properti ini untuk menentukan tujuan pemanggilan gagal dari sumber peristiwa MSK Amazon.

Jenis: DestinationConfig

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke DestinationConfig properti AWS::Lambda::EventSourceMapping sumber daya.

FilterCriteria

Objek yang mendefinisikan kriteria yang menentukan apakah Lambda harus memproses suatu peristiwa. Untuk informasi selengkapnya, lihat pemfilteran AWS Lambda acara di Panduan AWS Lambda Pengembang.

Jenis: FilterCriteria

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke FilterCriteria properti AWS::Lambda::EventSourceMapping sumber daya.

KmsKeyArn

Nama Sumber Daya Amazon (ARN) dari kunci untuk mengenkripsi informasi yang terkait dengan acara ini.

Tipe: String

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke KmsKeyArn properti AWS::Lambda::EventSourceMapping sumber daya.

MaximumBatchingWindowInSeconds

Jumlah waktu maksimum untuk mengumpulkan rekaman sebelum memanggil fungsi, dalam hitungan detik.

Tipe: Integer

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke MaximumBatchingWindowInSeconds properti AWS::Lambda::EventSourceMapping sumber daya.

ProvisionedPollerConfig

Konfigurasi untuk meningkatkan jumlah poller yang digunakan untuk menghitung pemetaan sumber peristiwa. Konfigurasi ini memungkinkan minimal 1 poller dan maksimal 20 poller. Sebagai contoh, lihatProvisionedPollerConfig contoh.

Jenis: ProvisionedPollerConfig

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke ProvisionedPollerConfig properti AWS::Lambda::EventSourceMapping sumber daya.

SchemaRegistryConfig

Konfigurasi untuk menggunakan registri skema dengan sumber acara Kafka.

catatan

Fitur ini ProvisionedPollerConfig perlu dikonfigurasi.

Jenis: SchemaRegistryConfig

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke AmazonManagedKafkaEventSourceConfig properti AWS::Lambda::EventSourceMapping sumber daya.

SourceAccessConfigurations

Susunan protokol autentikasi, komponen VPC, atau host virtual untuk mengamankan dan menentukan sumber peristiwa Anda.

Nilai yang valid: CLIENT_CERTIFICATE_TLS_AUTH

Jenis: Daftar SourceAccessConfiguration

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini adalah bagian dari AmazonManagedKafkaEventSourceConfigproperti sumber daya. AWS::Lambda::EventSourceMapping

StartingPosition

Posisi dalam pengaliran tempat untuk mulai membaca.

AT_TIMESTAMP— Tentukan waktu untuk mulai membaca catatan.
LATEST— Baca hanya catatan baru.
TRIM_HORIZON— Memproses semua catatan yang tersedia.

Nilai yang valid: AT_TIMESTAMP | LATEST | TRIM_HORIZON

Tipe: String

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke StartingPosition properti AWS::Lambda::EventSourceMapping sumber daya.

StartingPositionTimestamp

Waktu untuk mulai membaca, dalam detik waktu Unix. Tentukan StartingPositionTimestamp kapan StartingPosition ditentukan sebagaiAT_TIMESTAMP.

Tipe: Ganda

Wajib: Tidak

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke StartingPositionTimestamp properti AWS::Lambda::EventSourceMapping sumber daya.

Stream

Amazon Resource Name (ARN) dari aliran data atau konsumen pengaliran.

Tipe: String

Wajib: Ya

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke EventSourceArn properti AWS::Lambda::EventSourceMapping sumber daya.

Topics

Nama topik Kafka.

Tipe: Daftar

Wajib: Ya

AWS CloudFormation kompatibilitas: Properti ini diteruskan langsung ke Topics properti AWS::Lambda::EventSourceMapping sumber daya.

Contoh

Penyiapan lengkap dengan peran IAM

Contoh berikut menunjukkan penyiapan lengkap termasuk konfigurasi peran IAM yang diperlukan untuk menggunakan Schema Registry:


Parameters:
  PreCreatedSubnetOne:
    Type: String
  PreCreatedSubnetTwo:
    Type: String
  MskClusterName4:
    Type: String
 
Resources:
  MyLambdaExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Action: [sts:AssumeRole]
          Effect: Allow
          Principal:
            Service: [lambda.amazonaws.com]
      Policies:
      - PolicyName: KafkaClusterPermissions
        PolicyDocument:
          Statement:
          - Action: [kafka:DescribeClusterV2, kafka:GetBootstrapBrokers]
            Effect: Allow
            Resource: 'arn:aws:kafka:us-east-1:123456789012:cluster/*'
      - PolicyName: KafkaAuthPolicy
        PolicyDocument:
          Statement:
          - Action: [secretsmanager:GetSecretValue, kms:Decrypt]
            Effect: "Allow"            
            Resource: ['arn:aws:secretsmanager:us-west-2:123456789012:secret:kafkaSecret-******',
                        'arn:aws:kms:us-west-2:123456789012:key/keyId']
      - PolicyName: ENIPolicy
        PolicyDocument:
          Statement:
          - Action: [ec2:CreateNetworkInterface,
              ec2:DescribeNetworkInterfaces, ec2:DescribeVpcs, ec2:DeleteNetworkInterface,
              ec2:DescribeSubnets, ec2:DescribeSecurityGroups]
            Effect: Allow
            Resource: '*'      
      - PolicyName: SchemaRegistryPolicy
        PolicyDocument:
          Statement:
          - Action: [glue:GetRegistry]
            Effect: Allow
            Resource: 'arn:aws:glue:{region}:{account-id}:registry/{registry-name}'
      - PolicyName: SchemaVersionsPolicy
        PolicyDocument:
          Statement:
          - Action: [glue:GetSchemaVersions]
            Effect: Allow
            Resource: '*'          
      ManagedPolicyArns:
      - !Sub arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Tags:
      - {Value: SAM, Key: lambda:createdBy}
 
  MyMskCluster:
    Type: AWS::MSK::Cluster
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
        - Ref: PreCreatedSubnetOne
        - Ref: PreCreatedSubnetTwo
        InstanceType: kafka.t3.small
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 1
      ClusterName:
        Ref: MskClusterName4
      KafkaVersion: 3.8.x
      NumberOfBrokerNodes: 2
 
  MyMskStreamProcessor:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: nodejs18.x
      Handler: index.handler
      CodeUri: ${codeuri}
      Role:
        Fn::GetAtt: [MyLambdaExecutionRole, Arn]
      Events:
        MyMskEvent:
          Type: MSK
          Properties:
            StartingPosition: LATEST
            Stream:
              Ref: MyMskCluster
            SourceAccessConfigurations:
            - Type: SASL_SCRAM_512_AUTH 
              URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
            Topics:
            - SchemaRegistryTestTopic
            ProvisionedPollerConfig:
              MinimumPollers: 1
            SchemaRegistryConfig:
              AccessConfigs:
              - Type: BASIC_AUTH
                URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
              SchemaValidationConfigs:
              - Attribute: KEY
              EventRecordFormat: JSON
              SchemaRegistryURI: !Sub arn:${AWS::Partition}:glue:us-west-2:123456789012:registry/myregistry

ProvisionedPollerConfig contoh


ProvisionedPollerConfig:
  MinimumPollers: 1
  MaximumPollers: 20

Contoh Amazon MSK untuk Klaster yang ada

Berikut ini adalah contoh jenis sumber MSK peristiwa untuk cluster MSK Amazon yang sudah ada di file Akun AWS.

YAML


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream: arn:aws:kafka:us-east-1:012345678012:cluster/exampleClusterName/abcdefab-1234-abcd-5678-cdef0123ab01-2
      Topics:
        - MyTopic

Amazon MSK contoh untuk Klaster yang dinyatakan dalam Templat yang sama

Berikut ini adalah contoh tipe sumber peristiwa MSK untuk klaster Amazon MSK yang dinyatakan dalam file templat yang sama.

YAML


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream:
        Ref: MyMskCluster   # This must be the name of an MSK cluster declared in the same template file
      Topics:
        - MyTopic

Sumber Acara MSK dengan Registri Skema

Berikut ini adalah contoh dari jenis sumber MSK peristiwa dikonfigurasi dengan registri skema.


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream:
        Ref: MyMskCluster
      Topics:
        - SchemaRegistryTestTopic
      ProvisionedPollerConfig:
        MinimumPollers: 1
      SchemaRegistryConfig:
        SchemaRegistryURI: !Sub arn:${AWS::Partition}:glue:us-west-2:123456789012:registry/myregistry
        EventRecordFormat: JSON
        SchemaValidationConfigs:
          - Attribute: KEY
          - Attribute: VALUE

Sumber Acara MSK dengan Registri Skema Konfluen

Berikut ini adalah contoh dari jenis sumber MSK peristiwa dikonfigurasi dengan Confluent Schema Registry.


Events:
  MSKEvent:
    Type: MSK
    Properties:
      StartingPosition: LATEST
      Stream:
        Ref: MyMskCluster
      Topics:
        - SchemaRegistryTestTopic
      ProvisionedPollerConfig:
        MinimumPollers: 1
      SchemaRegistryConfig:
        SchemaRegistryURI: https://my-schema-registry.confluent.cloud
        AccessConfigs:
          - Type: BASIC_AUTH
            URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-secret
        EventRecordFormat: JSON
        SchemaValidationConfigs:
          - Attribute: KEY
          - Attribute: VALUE

Awas Javascript dinonaktifkan atau tidak tersedia di browser Anda.

Untuk menggunakan Dokumentasi AWS, Javascript harus diaktifkan. Lihat halaman Bantuan browser Anda untuk petunjuk.

Konvensi Dokumen