Menelan data streaming menggunakan Kinesis

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.

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"
}

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.

Buat tampilan terwujud untuk menggunakan data aliran. Contoh berikut menunjukkan kedua metode mendefinisikan tampilan terwujud untuk menelan data sumber JSON.

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" ;

Sebaliknya, dalam definisi tampilan terwujud berikut, tampilan terwujud memiliki skema yang ditentukan dalam Pergeseran Merah. Tampilan terwujud didistribusikan pada nilai UUID dari aliran dan diurutkan berdasarkan nilainya. approximatearrivaltimestamp

CREATE MATERIALIZED VIEW ev_station_data_extract DISTKEY(6) sortkey(1) AUTO REFRESH YES AS
    SELECT refresh_time,
    approximate_arrival_timestamp,
    partition_key,
    shard_id,
    sequence_number,
    json_extract_path_text(from_varbyte(kinesis_data,'utf-8'),'_id',true)::character(36) as ID,
    json_extract_path_text(from_varbyte(kinesis_data,'utf-8'),'clusterID',true)::varchar(30) as clusterID,
    json_extract_path_text(from_varbyte(kinesis_data,'utf-8'),'connectionTime',true)::varchar(20) as connectionTime,
    json_extract_path_text(from_varbyte(kinesis_data,'utf-8'),'kWhDelivered',true)::DECIMAL(10,2) as kWhDelivered,
    json_extract_path_text(from_varbyte(kinesis_data,'utf-8'),'stationID',true)::DECIMAL(10,2) as stationID,
    json_extract_path_text(from_varbyte(kinesis_data,'utf-8'),'spaceID',true)::varchar(100) as spaceID,
    json_extract_path_text(from_varbyte(kinesis_data, 'utf-8'),'timezone',true)::varchar(30)as timezone,
    json_extract_path_text(from_varbyte(kinesis_data,'utf-8'),'userID',true)::varchar(30) as userID
    FROM evdata."ev_station_data"
    WHERE LENGTH(kinesis_data) < 65355;

Kueri tampilan terwujud yang diperbarui untuk mendapatkan statistik penggunaan.

SELECT to_timestamp(connectionTime, 'YYYY-MM-DD HH24:MI:SS') as connectiontime
,SUM(kWhDelivered) AS Energy_Consumed 
,count(distinct userID) AS #Users
from ev_station_data_extract
group by  to_timestamp(connectionTime, 'YYYY-MM-DD HH24:MI:SS')
order by 1 desc;

Lihat hasil.

connectiontime	        energy_consumed	#users
2022-02-08 16:07:21+00	  4139	        10
2022-02-08 16:07:20+00	  5571	        10
2022-02-08 16:07:19+00	  8697	        20
2022-02-08 16:07:18+00	  4408	        10
2022-02-08 16:07:17+00	  4257	        10
2022-02-08 16:07:16+00	  6861	        10
2022-02-08 16:07:15+00	  5643	        10
2022-02-08 16:07:14+00	  3677	        10
2022-02-08 16:07:13+00	  4673	        10
2022-02-08 16:07:11+00	  9689	        20