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

Menggunakan pemfilteran peristiwa dengan sumber peristiwa Kinesis

Anda dapat menggunakan pemfilteran peristiwa untuk mengontrol rekaman mana dari aliran atau antrian yang dikirim Lambda ke fungsi Anda. Untuk informasi umum tentang cara kerja pemfilteran peristiwa, lihatKontrol peristiwa mana yang dikirim Lambda ke fungsi Anda.

Bagian ini berfokus pada pemfilteran peristiwa untuk sumber peristiwa Kinesis.

Dasar-dasar pemfilteran peristiwa Kinesis

Misalkan produsen menempatkan data berformat JSON ke dalam aliran data Kinesis Anda. Contoh catatan akan terlihat seperti berikut, dengan data JSON dikonversi ke string yang dikodekan Base64 di data lapangan.

{
    "kinesis": {
        "kinesisSchemaVersion": "1.0",
        "partitionKey": "1",
        "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
        "data": "eyJSZWNvcmROdW1iZXIiOiAiMDAwMSIsICJUaW1lU3RhbXAiOiAieXl5eS1tbS1kZFRoaDptbTpzcyIsICJSZXF1ZXN0Q29kZSI6ICJBQUFBIn0=",
        "approximateArrivalTimestamp": 1545084650.987
        },
    "eventSource": "aws:kinesis",
    "eventVersion": "1.0",
    "eventID": "shardId-000000000006:49590338271490256608559692538361571095921575989136588898",
    "eventName": "aws:kinesis:record",
    "invokeIdentityArn": "arn:aws:iam::123456789012:role/lambda-role",
    "awsRegion": "us-east-2",
    "eventSourceARN": "arn:aws:kinesis:us-east-2:123456789012:stream/lambda-stream"
}

Selama data yang dimasukkan produsen ke dalam aliran adalah JSON yang valid, Anda dapat menggunakan pemfilteran peristiwa untuk memfilter catatan menggunakan data kunci. Misalkan produser memasukkan catatan ke aliran Kinesis Anda dalam format JSON berikut.

{
    "record": 12345,
    "order": {
        "type": "buy",
        "stock": "ANYCO",
        "quantity": 1000
        }
}

Untuk memfilter hanya catatan yang tipe pesanan adalah “beli,” FilterCriteria objeknya adalah sebagai berikut.

{
    "Filters": [
        {
            "Pattern": "{ \"data\" : { \"order\" : { \"type\" : [ \"buy\" ] } } }"
        }
    ]
}

Untuk kejelasan tambahan, berikut adalah nilai filter yang Pattern diperluas dalam JSON biasa.

{
    "data": {
        "order": {
            "type": [ "buy" ]
            }
      }
}

Anda dapat menambahkan filter menggunakan konsol, AWS CLI atau AWS SAM templat.

Console

Untuk menambahkan filter ini menggunakan konsol, ikuti instruksi di Melampirkan kriteria filter ke pemetaan sumber peristiwa (konsol) dan masukkan string berikut untuk kriteria Filter.

{ "data" : { "order" : { "type" : [ "buy" ] } } }

AWS CLI

Untuk membuat pemetaan sumber peristiwa baru dengan kriteria filter ini menggunakan AWS Command Line Interface (AWS CLI), jalankan perintah berikut.

aws lambda create-event-source-mapping \
    --function-name my-function \
    --event-source-arn arn:aws:kinesis:us-east-2:123456789012:stream/my-stream \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : { \"order\" : { \"type\" : [ \"buy\" ] } } }"}]}'

Untuk menambahkan kriteria filter ini ke pemetaan sumber peristiwa yang ada, jalankan perintah berikut.

aws lambda update-event-source-mapping \
    --uuid "a1b2c3d4-5678-90ab-cdef-11111EXAMPLE" \
    --filter-criteria '{"Filters": [{"Pattern": "{ \"data\" : { \"order\" : { \"type\" : [ \"buy\" ] } } }"}]}'

AWS SAM

Untuk menambahkan filter ini menggunakan AWS SAM, tambahkan cuplikan berikut ke template YAML untuk sumber acara Anda.

FilterCriteria:
  Filters:
    - Pattern: '{ "data" : { "order" : { "type" : [ "buy" ] } } }'

Untuk memfilter peristiwa dari sumber Kinesis dengan benar, bidang data dan kriteria filter Anda untuk bidang data harus dalam format JSON yang valid. Jika salah satu bidang tidak dalam format JSON yang valid, Lambda menghapus pesan atau melempar pengecualian. Tabel berikut merangkum perilaku spesifik:

Memfilter rekaman agregat Kinesis

Dengan Kinesis, Anda dapat menggabungkan beberapa catatan ke dalam satu catatan Kinesis Data Streams untuk meningkatkan throughput data Anda. Lambda hanya dapat menerapkan kriteria filter ke rekaman agregat saat Anda menggunakan Kinesis yang disempur nakan fan-out. Memfilter rekaman agregat dengan Kinesis standar tidak didukung. Saat menggunakan fan-out yang disempurnakan, Anda mengonfigurasi konsumen throughput khusus Kinesis untuk bertindak sebagai pemicu fungsi Lambda Anda. Lambda kemudian menyaring catatan agregat dan meneruskan hanya catatan yang memenuhi kriteria filter Anda.

Untuk mempelajari lebih lanjut tentang agregasi rekaman Kinesis, lihat bagian A gregasi di halaman Konsep Utama Kinesis Producer Library (KPL). Untuk Mempelajari lebih lanjut tentang menggunakan Lambda dengan fan-out yang disempurnakan Kinesis, lihat Meningkatkan kinerja pemrosesan aliran real-time dengan Amazon Kinesis Data Streams yang disempurnakan fan-out dan Lambda di blog komputasi. AWS AWS