Ниже приведен вариант использования, над которым я работаю: я настроил enable Streams при создании DynamoDB с new and old Image. Я создал Kinesis Firehose delivery stream с пунктом назначения как Redshift(Intermediate s3).
Из Dynamodb мой поток достигает Firhose и оттуда в Bucket в формате JSON (S3 Bucket -Gzip), приведенном ниже. Моя проблема в том, что я cannot COPY this JSON to redshift.
Вещи, которые я не могу получить:
- Not Sure what should be the Create table Statement in Redshift
- What should be the COPY Syntax in Kinesis firhose.
- How should i use JsonPaths here. Kinesis Data firehouse set to return only json to my s3 bucket.
- How to mention the Maniphest in the COPY Command
Загрузка JSON в S3 показана ниже:
{
"Keys": {
"vehicle_id": {
"S": "x011"
}
},
"NewImage": {
"heart_beat": {
"N": "0"
},
"cdc_id": {
"N": "456"
},
"latitude": {
"N": "1.30951"
},
"not_deployed_counter": {
"N": "1"
},
"reg_ind": {
"N": "0"
},
"operator": {
"S": "x"
},
"d_dttm": {
"S": "11/08/2018 2:43:46 PM"
},
"z_id": {
"N": "1267"
},
"last_end_trip_dttm": {
"S": "11/08/2018 1:43:46 PM"
},
"land_ind": {
"N": "1"
},
"s_ind": {
"N": "1"
},
"status_change_dttm": {
"S": "11/08/2018 2:43:46 PM"
},
"case_ind": {
"N": "1"
},
"last_po_change_dttm": {
"S": "11/08/2018 2:43:46 PM"
},
"violated_duration": {
"N": "20"
},
"vehicle_id": {
"S": "x011"
},
"longitude": {
"N": "103.7818"
},
"file_status": {
"S": "Trip_Start"
},
"unhired_duration": {
"N": "10"
},
"eo_lat": {
"N": "1.2345"
},
"reply_eo_ind": {
"N": "1"
},
"license_ind": {
"N": "0"
},
"indiscriminately_parked_ind": {
"N": "0"
},
"eo_lng": {
"N": "102.8978"
},
"officer_id": {
"S": "[email protected]"
},
"case_status": {
"N": "0"
},
"color_status_cd": {
"N": "0"
},
"parking_id": {
"N": "2345"
},
"ttr_dttm": {
"S": "11/08/2018 2:43:46 PM"
},
"deployed_ind": {
"N": "1"
},
"status": {
"S": "PI"
}
},
"SequenceNumber": "1200000000000956615967",
"SizeBytes": 570,
"ApproximateCreationDateTime": 1535513040,
"eventName": "INSERT"
}
Мое заявление о создании таблицы:
create table vehicle_status(
heart_beat integer,
cdc_id integer,
latitude integer,
not_deployed_counter integer,
reg_ind integer,
operator varchar(10),
d_dttm varchar(30),
z_id integer,
last_end_trip_dttm varchar(30),
land_ind integer,
s_ind integer,
status_change_dttm varchar(30),
case_ind integer,
last_po_change_dttm varchar(30),
violated_duration integer,
vehicle_id varchar(8),
longitude integer,
file_status varchar(30),
unhired_duration integer,
eo_lat integer,
reply_eo_ind integer,
license_ind integer,
indiscriminately_parked_ind integer,
eo_lng integer,
officer_id varchar(50),
case_status integer,
color_status_cd integer,
parking_id integer,
ttr_dttm varchar(30),
deployed_ind varchar(3),
status varchar(8));
И мое заявление о копировании (вручную пытаюсь повторить это из Redshift):
COPY vehicle_status (heart_beat, cdc_id, latitude, not_deployed_counter, reg_ind, operator, d_dttm, z_id, last_end_trip_dttm, land_ind, s_ind, status_change_dttm, case_ind, last_po_change_dttm, violated_duration, vehicle_id, longitude, file_status, unhired_duration, eo_lat, reply_eo_ind, license_ind, indiscriminately_parked_ind, eo_lng, officer_id, case_status, color_status_cd, parking_id, ttr_dttm, deployed_ind, status)
FROM 's3://<my-bucket>/2018/08/29/05/vehicle_status_change-2-2018-08-29-05-24-42-092c330b-e14a-4133-bf4a-5982f2e1f49e.gz' CREDENTIALS 'aws_iam_role=arn:aws:iam::<accountnum>:role/<RedshiftRole>' GZIP json 'auto';
Когда я пытаюсь выполнить описанную выше процедуру, я получаю команду «Вставить записи», но все столбцы и строки пусты.
Как я могу скопировать этот формат json в redhsift. Застряли здесь последние 3 дня. Любая помощь по этому поводу подойдет.
Ведро S3:
Amazon S3/<My-bucket>/2018/08/29/05
Amazon S3/<My-bucket>/manifests/2018/08/29/05
