Menelan data streaming menggunakan Kinesis - Amazon Redshift

Amazon Redshift tidak akan lagi mendukung pembuatan Python UDFs baru mulai Patch 198. Python yang ada UDFs akan terus berfungsi hingga 30 Juni 2026. Untuk informasi lebih lanjut, lihat posting blog.

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

Menelan data streaming menggunakan Kinesis

Prosedur ini menunjukkan cara menelan data dari aliran Kinesis bernama ev_station_data, yang berisi data konsumsi dari stasiun pengisian EV yang berbeda, dalam format JSON. Skema didefinisikan dengan baik. Contoh menunjukkan cara menyimpan data sebagai JSON mentah dan juga cara mengonversi data JSON ke tipe data Amazon Redshift saat dicerna.

Pengaturan produser

  1. Menggunakan Amazon Kinesis Data Streams, ikuti langkah-langkah untuk membuat aliran bernama. ev_station_data Pilih Sesuai Permintaan untuk mode Kapasitas. Untuk informasi selengkapnya, lihat Membuat Stream melalui Konsol AWS Manajemen.

  2. Amazon Kinesis Data Generator dapat membantu Anda menghasilkan data pengujian untuk digunakan dengan streaming Anda. Ikuti langkah-langkah yang dijelaskan dalam alat untuk memulai, dan gunakan templat data berikut untuk menghasilkan data Anda:

    {
    
   "_id" : "{{random.uuid}}",
   "clusterID": "{{random.number(
        {   "min":1,
            "max":50
        }
    )}}", 
    "connectionTime": "{{date.now("YYYY-MM-DD HH:mm:ss")}}",
    "kWhDelivered": "{{commerce.price}}",
    "stationID": "{{random.number(
        {   "min":1,
            "max":467
        }
    )}}",
      "spaceID": "{{random.word}}-{{random.number(
        {   "min":1,
            "max":20
        }
    )}}",
 
   "timezone": "America/Los_Angeles",
   "userID": "{{random.number(
        {   "min":1000,
            "max":500000
        }
    )}}"
}

    Setiap objek JSON dalam data aliran memiliki properti berikut: 

    {
    "_id": "12084f2f-fc41-41fb-a218-8cc1ac6146eb",
    "clusterID": "49",
    "connectionTime": "2022-01-31 13:17:15",
    "kWhDelivered": "74.00",
    "stationID": "421",
    "spaceID": "technologies-2",
    "timezone": "America/Los_Angeles",
    "userID": "482329"
}

Pengaturan Amazon Redshift

Langkah-langkah ini menunjukkan cara mengonfigurasi tampilan terwujud untuk mencerna data.

  1. Buat skema eksternal untuk memetakan data dari Kinesis ke objek Redshift.

    CREATE EXTERNAL SCHEMA evdata FROM KINESIS
IAM_ROLE 'arn:aws:iam::0123456789:role/redshift-streaming-role';

    Untuk informasi tentang cara mengonfigurasi peran IAM, lihatMemulai dengan konsumsi streaming dari Amazon Kinesis Data Streams.

  2. Buat tampilan terwujud untuk menggunakan data aliran. Contoh berikut menunjukkan bagaimana mendefinisikan tampilan terwujud untuk menelan data yang diformat JSON dari aliran Kinesis.

    Pertama, simpan catatan aliran dalam format SUPER semi-terstruktur. Dalam contoh ini, sumber JSON disimpan dalam Redshift tanpa mengonversi ke tipe Redshift.

    CREATE MATERIALIZED VIEW ev_station_data AS
    SELECT approximate_arrival_timestamp,
    partition_key,
    shard_id,
    sequence_number,
    case when can_json_parse(kinesis_data) then json_parse(kinesis_data) else null end as payload,
    case when not can_json_parse(kinesis_data) then kinesis_data else null end as failed_payload
    FROM evdata."ev_station_data" ;

Kueri aliran

  1. Aktifkan atribut SUPER peka huruf besar/kecil menggunakan perintah di bawah ini. Amazon Redshift tidak peka huruf besar/kecil secara default, jadi untuk mengakses atribut SUPER yang peka huruf besar/kecil, Anda harus mengaktifkan fungsi ini.

    SET enable_case_sensitive_super_attribute to TRUE;

  2. Segarkan tampilan terwujud dengan perintah berikut untuk menarik data dari aliran.

    REFRESH MATERIALIZED VIEW ev_station_data;

  3. Kueri tampilan terwujud yang diperbarui untuk mendapatkan statistik penggunaan.

    SELECT e.payload.connectionTime::date as connectiontime
,SUM(e.payload.kWhDelivered::decimal(10,2)) AS Energy_Consumed
,count(distinct e.payload.userID) AS #Users
from ev_station_data as e
group by connectiontime
order by 1 desc;

  4. Lihat hasil.

    connectiontime        energy_consumed    #users
2022-02-08                 4139          10
2022-02-09                 5571          10
2022-02-10                 8697          20
2022-02-11                 4408          10
2022-02-12                 4257          10
2022-02-23                 6861          10