amazon-s3 - 无法将 json - Dynamo db Streams 复制到 redshift

Question

以下是我正在研究的用例:我在使用新旧图像创建DynamoDB时配置了启用流。我创建了一个Kinesis Firehose传输流目标为 Redshift(Intermediate s3)。

从 Dynamodb，我的流到达 Firhose，并从那里以 JSON (S3 Bucket -Gzip) 形式到达存储桶，如下所示。我的问题是我无法将此 JSON 复制到 redshift。

我无法得到的东西:

• 1. 不确定 Redshift 中的创建表语句应该是什么
• Kinesis firhose 中的 COPY 语法应该是什么。
• 我应该如何在这里使用 JsonPaths。 Kinesis Data firehouse 设置为仅将 json 返回到我的 s3 存储桶。
• 如何在 COPY 命令中提及 Maniphest

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": "xxxx@gmail.com"
        },
        "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

Answer 1

我对亚马逊不太熟悉，但让我尝试回答您的大部分问题，以便您可以继续前进。非常欢迎其他人编辑此答案或其他详细信息。谢谢!

Not Sure what should be the Create table Statement in Redshift

您的创建语句create table vehicle_status(...)没有问题，不过你可以添加 distribution key , sort key和encoding根据您的要求，引用更多here和 here

根据 AWS Kenesis documents ，您的表必须出现在 Redshift 中，因此您可以连接到 Redshift使用psql命令并运行 create statement手动。

What should be the COPY Syntax in Kinesis firhose.

Copy无论您通过 psql 运行它，语法都将保持不变或firhose ，幸运的是，您提出的复制脚本可以正常工作，没有任何错误，我在我的实例中尝试了直接 AWS/SECRET 的小修改关键供应而不是它工作正常，这里是 sql我运行得很好，并将 1 条数据记录复制到表 vehicle_status .

实际上你的json路径结构很复杂，因此json 'auto'不管用。这是工作命令，我创建了一个示例 jsonpath文件包含 4 个示例字段，您可以遵循相同的结构来创建 jsonpath包含所有数据点的文件。

 COPY vehicle_status (heart_beat, cdc_id, operator, status) FROM 's3://XXX/development/test_file.json' CREDENTIALS 'aws_access_key_id=XXXXXXXXXXXXXXXXX;aws_secret_access_key=MYXXXXXXXXXXXXXXXXXXXXXX' json 's3://XXX/development/yourjsonpathfile';

还有你的json path file应具有类似于以下内容。

{
  "jsonpaths": [
    "$['NewImage']['heart_beat']['N']",
    "$['NewImage']['cdc_id']['N']",
    "$['NewImage']['operator']['S']",
    "$['NewImage']['status']['S']"
  ]
}

我已经测试过它并且有效。

How should i use JsonPaths here. Kinesis Data firehouse set to return only json to my s3 bucket.

我用了你的例子json仅数据且有效，因此我认为这里没有问题。

How to mention the Maniphest in the COPY Command

这是个好问题，我可以尝试解释一下，希望您在这里指的是menifest .

如果你看到上面的复制命令，它对于一个或几个文件来说效果很好，但是如果你有很多文件，这里就出现了 menifest 的概念。 。直接来自 Amazon 文档，“您无需提供 COPY 命令的对象路径，而是提供显式列出要加载的文件的 JSON 格式文本文件的名称。”

简而言之，如果您想一次加载多个文件，这也是首选方式 Redshift ，您可以创建一个简单的 menifest使用 json 并在复制命令中提供相同的内容。

{
"entries": [
{"url":"s3://mybucket-alpha/2013-10-04-custdata", "mandatory":true},
{"url":"s3://mybucket-alpha/2013-10-05-custdata", "mandatory":true},....
]
}

上传菜单至S3并在复制命令中使用相同的内容，如下所示。

 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://XXX/development/test.menifest' CREDENTIALS 'aws_access_key_id=XXXXXXXXXXXXXXXXX;aws_secret_access_key=MYXXXXXXXXXXXXXXXXXXXXXX' json 's3://yourbucket/jsonpath' menifest;

这里是 menifest 的详细引用.

我希望这能给您一些想法，如何继续，如果您看到特定错误，我很乐意重新关注答案。